def loseConnectionWithProducer(self, writeBlockedOnRead):
        clientProtocol, tlsProtocol, producer = self.setupStreamingProducer()
        serverProtocol, serverTLSProtocol = buildTLSProtocol(server=True)

        if not writeBlockedOnRead:
            # Do the initial handshake before write:
            self.flushTwoTLSProtocols(tlsProtocol, serverTLSProtocol)
        else:
            # In this case the write below will trigger write-blocked-on-read
            # condition...
            pass

        # Now write, then lose connection:
        clientProtocol.transport.write("x ")
        clientProtocol.transport.loseConnection()
        self.flushTwoTLSProtocols(tlsProtocol, serverTLSProtocol)

        # Underlying transport should not have loseConnection called yet, nor
        # should producer be stopped:
        self.assertEqual(tlsProtocol.transport.disconnecting, False)
        self.assertFalse("stop" in producer.producerHistory)

        # Writes from client to server should continue to go through, since we
        # haven't unregistered producer yet:
        clientProtocol.transport.write("hello")
        clientProtocol.transport.writeSequence([" ", "world"])

        # Unregister producer; this should trigger TLS shutdown:
        clientProtocol.transport.unregisterProducer()
        self.assertNotEqual(tlsProtocol.transport.value(), "")
        self.assertEqual(tlsProtocol.transport.disconnecting, False)

        # Additional writes should not go through:
        clientProtocol.transport.write("won't")
        clientProtocol.transport.writeSequence(["won't!"])

        # Finish TLS close handshake:
        self.flushTwoTLSProtocols(tlsProtocol, serverTLSProtocol)
        self.assertEqual(tlsProtocol.transport.disconnecting, True)

        # Bytes made it through, as long as they were written before producer
        # was unregistered:
        self.assertEqual("".join(serverProtocol.received), "x hello world")
Common code for tests involving writes by producer after
dataset/ETHPy150Open nlloyd/SubliminalCollaborator/libs/twisted/protocols/test/test_tls.py/TLSProducerTests.loseConnectionWithProducer