Twisted Network Programming Skill

Core Architecture

Reactor ← event loop: polls OS for I/O events, timer events, network events
Transport ← connection abstraction: write(), loseConnection(), peer info
Protocol ← event handler: dataReceived(), connectionMade(), connectionLost()
Factory ← creates Protocol instances per connection: buildProtocol()

---

TCP Echo Server and Client

# Server
from twisted.internet import protocol, reactor

class Echo(protocol.Protocol):
    def dataReceived(self, data):
        self.transport.write(data)   # echo back

class EchoFactory(protocol.Factory):
    def buildProtocol(self, addr):
        return Echo()

reactor.listenTCP(8000, EchoFactory())
reactor.run()

# Client
from twisted.internet import reactor, protocol

class EchoClient(protocol.Protocol):
    def connectionMade(self):
        self.transport.write(b"Hello, world!")

    def dataReceived(self, data):
        print("Server said:", data)
        self.transport.loseConnection()

class EchoClientFactory(protocol.ClientFactory):
    def buildProtocol(self, addr):
        return EchoClient()

    def clientConnectionFailed(self, connector, reason):
        print("Connection failed:", reason)
        reactor.stop()

    def clientConnectionLost(self, connector, reason):
        reactor.stop()

reactor.connectTCP("localhost", 8000, EchoClientFactory())
reactor.run()

---

Protocol Patterns

Factory Shorthand (common idiom)

class MyFactory(protocol.Factory):
    protocol = MyProtocol      # auto-instantiates; buildProtocol not needed
    # persistent state stays in factory:
    def __init__(self):
        self.connections = {}  # shared across protocol instances

State Machine Protocol

from twisted.protocols.basic import LineReceiver

class AuthProtocol(LineReceiver):
    def __init__(self, factory):
        self.factory = factory
        self.state = "AUTHENTICATE"

    def connectionMade(self):
        self.sendLine(b"Username:")

    def lineReceived(self, line):
        handler = getattr(self, f"handle_{self.state}")
        handler(line.decode())

    def handle_AUTHENTICATE(self, username):
        if username in self.factory.users:
            self.state = "ACTIVE"
            self.sendLine(b"Welcome!")
        else:
            self.sendLine(b"Unknown user.")

    def handle_ACTIVE(self, message):
        self.sendLine(f"Echo: {message}".encode())

---

Deferreds (Async Promises)

Basic Callback/Errback Chain

from twisted.internet.defer import Deferred

def addBold(result):
    return f"<b>{result}</b>"

def addItalic(result):
    return f"<i>{result}</i>"

def printHTML(result):
    print(result)

def handleError(failure):
    print(f"Error: {failure}")

d = Deferred()
d.addCallback(addBold)      # callback chain: addBold → addItalic → printHTML
d.addCallback(addItalic)
d.addCallback(printHTML)
d.addErrback(handleError)   # errback chain fires on exception/errback()

d.callback("Hello World")   # fires success chain → "<i><b>Hello World</b></i>"
# d.errback(Exception("...")) would fire error chain

Deferred Rules

1. Returning a value from a callback passes it to the next callback
2. Raising an exception in a callback switches to the errback chain
3. Returning a value from an errback switches back to the callback chain
4. Never block in a callback — it blocks the reactor (whole server)
5. Use addCallbacks(cb, eb) to register both at same level
6. reactor.callLater(seconds, func) schedules deferred events

Using Deferreds in the Reactor

from twisted.internet import reactor, defer

class ResourceFetcher:
    def fetch(self, url):
        d = defer.Deferred()
        # simulate async: in real code, use reactor-integrated HTTP client
        reactor.callLater(1, d.callback, "resource content")
        d.addCallback(self._process)
        return d

    def _process(self, content):
        return content.upper()

def printResult(result):
    print(result)
    reactor.stop()

f = ResourceFetcher()
d = f.fetch("http://example.com")
d.addCallback(printResult)
reactor.run()

Running Blocking Code (threads)

from twisted.internet import reactor, threads

def blocking_db_query():
    import time; time.sleep(2)   # simulated blocking op
    return "result"

d = threads.deferToThread(blocking_db_query)
d.addCallback(lambda r: print("Got:", r))
reactor.run()

---

Twisted Web (HTTP Server)

from twisted.web import server, resource
from twisted.internet import reactor

class HelloResource(resource.Resource):
    isLeaf = True                    # no child resources

    def render_GET(self, request):
        return b"Hello, world!"

    def render_POST(self, request):
        data = request.content.read()
        return b"Got: " + data

class Root(resource.Resource):
    def getChild(self, name, request):
        if name == b"hello":
            return HelloResource()
        return resource.NoResource()

    def render_GET(self, request):
        return b"Root"

reactor.listenTCP(8080, server.Site(Root()))
reactor.run()

---

Twisted Web Client (HTTP)

from twisted.web.client import Agent
from twisted.internet import reactor

agent = Agent(reactor)

def printResponse(response):
    print(response.code, response.headers)
    return response

def printError(failure):
    print("Error:", failure)

d = agent.request(b"GET", b"http://example.com")
d.addCallback(printResponse)
d.addErrback(printError)
d.addBoth(lambda _: reactor.stop())
reactor.run()

---

SSH Server (Conch)

from twisted.cred import portal, checkers
from twisted.conch import manhole, manhole_ssh
from twisted.internet import reactor

def getRealm():
    return manhole_ssh.TerminalRealm()

portal_obj = portal.Portal(getRealm())
portal_obj.registerChecker(
    checkers.InMemoryUsernamePasswordDatabaseDontUse(admin=b"password"))

factory = manhole_ssh.ConchFactory(portal_obj)
reactor.listenTCP(2222, factory)
reactor.run()
# Provides interactive Python shell over SSH

---

Testing with Trial

from twisted.trial import unittest
from twisted.test import proto_helpers

class EchoTestCase(unittest.TestCase):
    def setUp(self):
        factory = EchoFactory()
        self.proto = factory.buildProtocol(("127.0.0.1", 0))
        self.tr = proto_helpers.StringTransport()
        self.proto.makeConnection(self.tr)

    def test_echo(self):
        self.proto.dataReceived(b"Hello")
        self.assertEqual(self.tr.value(), b"Hello")

    def test_deferred(self):
        d = some_deferred_function()
        d.addCallback(self.assertEqual, "expected")
        return d   # Trial waits for the Deferred to fire

# Run tests
trial tests/   # or: trial mypackage.tests.TestClass

---

Deployment with twistd

# myservice.tac (Twisted Application Configuration)
from twisted.application import service, internet
from twisted.internet import protocol

class MyFactory(protocol.Factory):
    protocol = MyProtocol

application = service.Application("My Server")
tcp_service = internet.TCPServer(8000, MyFactory())
tcp_service.setServiceParent(application)

# Run as daemon (logs to twistd.log)
twistd -y myservice.tac

# Foreground (useful for development)
twistd --nodaemon -y myservice.tac

---

Event-Driven vs. Multithreaded vs. Synchronous

Model	Concurrency	Threads	Shared State	When to Use
Synchronous	None	1	No sync needed	Simple scripts
Multithreaded	OS threads	N	Requires locks/mutexes	CPU-bound
Event-driven (Twisted)	Single-threaded async	1 + deferToThread	No sync needed (in reactor)	I/O-bound, many connections

Key rule: Never block in the reactor thread. Use deferToThread() for blocking calls.