1 # -*- test-case-name: twisted.test.test_factories,twisted.internet.test.test_protocol -*-
2 # Copyright (c) Twisted Matrix Laboratories.
3 # See LICENSE for details.
6 Standard implementations of Twisted protocol-related interfaces.
8 Start here if you are looking to write a new protocol implementation for
9 Twisted. The Protocol class contains some introductory material.
11 Maintainer: Itamar Shtull-Trauring
15 from zope.interface import implements
18 from twisted.python import log, failure, components
19 from twisted.internet import interfaces, error, defer
24 This is a factory which produces protocols.
26 By default, buildProtocol will create a protocol of the class given in
30 implements(interfaces.IProtocolFactory, interfaces.ILoggingContext)
32 # put a subclass of Protocol here:
40 Describe this factory for log messages.
42 return self.__class__.__name__
46 """Make sure startFactory is called.
48 Users should not call this function themselves!
52 log.msg("Starting factory %r" % self)
54 self.numPorts = self.numPorts + 1
57 """Make sure stopFactory is called.
59 Users should not call this function themselves!
61 if self.numPorts == 0:
62 # this shouldn't happen, but does sometimes and this is better
63 # than blowing up in assert as we did previously.
65 self.numPorts = self.numPorts - 1
68 log.msg("Stopping factory %r" % self)
71 def startFactory(self):
72 """This will be called before I begin listening on a Port or Connector.
74 It will only be called once, even if the factory is connected
77 This can be used to perform 'unserialization' tasks that
78 are best put off until things are actually running, such
79 as connecting to a database, opening files, etcetera.
82 def stopFactory(self):
83 """This will be called before I stop listening on all Ports/Connectors.
85 This can be overridden to perform 'shutdown' tasks such as disconnecting
86 database connections, closing files, etc.
88 It will be called, for example, before an application shuts down,
89 if it was connected to a port. User code should not call this function
93 def buildProtocol(self, addr):
94 """Create an instance of a subclass of Protocol.
96 The returned instance will handle input on an incoming server
97 connection, and an attribute \"factory\" pointing to the creating
100 Override this method to alter how Protocol instances get created.
102 @param addr: an object implementing L{twisted.internet.interfaces.IAddress}
109 class ClientFactory(Factory):
110 """A Protocol factory for clients.
112 This can be used together with the various connectXXX methods in
116 def startedConnecting(self, connector):
117 """Called when a connection has been started.
119 You can call connector.stopConnecting() to stop the connection attempt.
121 @param connector: a Connector object.
124 def clientConnectionFailed(self, connector, reason):
125 """Called when a connection has failed to connect.
127 It may be useful to call connector.connect() - this will reconnect.
129 @type reason: L{twisted.python.failure.Failure}
132 def clientConnectionLost(self, connector, reason):
133 """Called when an established connection is lost.
135 It may be useful to call connector.connect() - this will reconnect.
137 @type reason: L{twisted.python.failure.Failure}
141 class _InstanceFactory(ClientFactory):
143 Factory used by ClientCreator.
145 @ivar deferred: The L{Deferred} which represents this connection attempt and
146 which will be fired when it succeeds or fails.
148 @ivar pending: After a connection attempt succeeds or fails, a delayed call
149 which will fire the L{Deferred} representing this connection attempt.
155 def __init__(self, reactor, instance, deferred):
156 self.reactor = reactor
157 self.instance = instance
158 self.deferred = deferred
162 return "<ClientCreator factory: %r>" % (self.instance, )
165 def buildProtocol(self, addr):
167 Return the pre-constructed protocol instance and arrange to fire the
168 waiting L{Deferred} to indicate success establishing the connection.
170 self.pending = self.reactor.callLater(
171 0, self.fire, self.deferred.callback, self.instance)
176 def clientConnectionFailed(self, connector, reason):
178 Arrange to fire the waiting L{Deferred} with the given failure to
179 indicate the connection could not be established.
181 self.pending = self.reactor.callLater(
182 0, self.fire, self.deferred.errback, reason)
186 def fire(self, func, value):
188 Clear C{self.pending} to avoid a reference cycle and then invoke func
198 Client connections that do not require a factory.
200 The various connect* methods create a protocol instance using the given
201 protocol class and arguments, and connect it, returning a Deferred of the
202 resulting protocol instance.
204 Useful for cases when we don't really need a factory. Mainly this
205 is when there is no shared state between protocol instances, and no need
208 The C{connectTCP}, C{connectUNIX}, and C{connectSSL} methods each return a
209 L{Deferred} which will fire with an instance of the protocol class passed to
210 L{ClientCreator.__init__}. These Deferred can be cancelled to abort the
211 connection attempt (in a very unlikely case, cancelling the Deferred may not
212 prevent the protocol from being instantiated and connected to a transport;
213 if this happens, it will be disconnected immediately afterwards and the
214 Deferred will still errback with L{CancelledError}).
217 def __init__(self, reactor, protocolClass, *args, **kwargs):
218 self.reactor = reactor
219 self.protocolClass = protocolClass
224 def _connect(self, method, *args, **kwargs):
226 Initiate a connection attempt.
228 @param method: A callable which will actually start the connection
229 attempt. For example, C{reactor.connectTCP}.
231 @param *args: Positional arguments to pass to C{method}, excluding the
234 @param **kwargs: Keyword arguments to pass to C{method}.
236 @return: A L{Deferred} which fires with an instance of the protocol
237 class passed to this L{ClientCreator}'s initializer or fails if the
238 connection cannot be set up for some reason.
240 def cancelConnect(deferred):
241 connector.disconnect()
242 if f.pending is not None:
244 d = defer.Deferred(cancelConnect)
245 f = _InstanceFactory(
246 self.reactor, self.protocolClass(*self.args, **self.kwargs), d)
247 connector = method(factory=f, *args, **kwargs)
251 def connectTCP(self, host, port, timeout=30, bindAddress=None):
253 Connect to a TCP server.
255 The parameters are all the same as to L{IReactorTCP.connectTCP} except
256 that the factory parameter is omitted.
258 @return: A L{Deferred} which fires with an instance of the protocol
259 class passed to this L{ClientCreator}'s initializer or fails if the
260 connection cannot be set up for some reason.
262 return self._connect(
263 self.reactor.connectTCP, host, port, timeout=timeout,
264 bindAddress=bindAddress)
267 def connectUNIX(self, address, timeout=30, checkPID=False):
269 Connect to a Unix socket.
271 The parameters are all the same as to L{IReactorUNIX.connectUNIX} except
272 that the factory parameter is omitted.
274 @return: A L{Deferred} which fires with an instance of the protocol
275 class passed to this L{ClientCreator}'s initializer or fails if the
276 connection cannot be set up for some reason.
278 return self._connect(
279 self.reactor.connectUNIX, address, timeout=timeout,
283 def connectSSL(self, host, port, contextFactory, timeout=30, bindAddress=None):
285 Connect to an SSL server.
287 The parameters are all the same as to L{IReactorSSL.connectSSL} except
288 that the factory parameter is omitted.
290 @return: A L{Deferred} which fires with an instance of the protocol
291 class passed to this L{ClientCreator}'s initializer or fails if the
292 connection cannot be set up for some reason.
294 return self._connect(
295 self.reactor.connectSSL, host, port,
296 contextFactory=contextFactory, timeout=timeout,
297 bindAddress=bindAddress)
301 class ReconnectingClientFactory(ClientFactory):
303 Factory which auto-reconnects clients with an exponential back-off.
305 Note that clients should call my resetDelay method after they have
306 connected successfully.
308 @ivar maxDelay: Maximum number of seconds between connection attempts.
309 @ivar initialDelay: Delay for the first reconnection attempt.
310 @ivar factor: A multiplicitive factor by which the delay grows
311 @ivar jitter: Percentage of randomness to introduce into the delay length
312 to prevent stampeding.
313 @ivar clock: The clock used to schedule reconnection. It's mainly useful to
314 be parametrized in tests. If the factory is serialized, this attribute
315 will not be serialized, and the default value (the reactor) will be
316 restored when deserialized.
317 @type clock: L{IReactorTime}
318 @ivar maxRetries: Maximum number of consecutive unsuccessful connection
319 attempts, after which no further connection attempts will be made. If
320 this is not explicitly set, no maximum is applied.
324 # Note: These highly sensitive factors have been precisely measured by
325 # the National Institute of Science and Technology. Take extreme care
326 # in altering them, or you may damage your Internet!
327 # (Seriously: <http://physics.nist.gov/cuu/Constants/index.html>)
328 factor = 2.7182818284590451 # (math.e)
329 # Phi = 1.6180339887498948 # (Phi is acceptable for use as a
330 # factor if e is too large for your application.)
331 jitter = 0.11962656472 # molar Planck constant times c, joule meter/mole
343 def clientConnectionFailed(self, connector, reason):
344 if self.continueTrying:
345 self.connector = connector
349 def clientConnectionLost(self, connector, unused_reason):
350 if self.continueTrying:
351 self.connector = connector
355 def retry(self, connector=None):
357 Have this connector connect again, after a suitable delay.
359 if not self.continueTrying:
361 log.msg("Abandoning %s on explicit request" % (connector,))
364 if connector is None:
365 if self.connector is None:
366 raise ValueError("no connector to retry")
368 connector = self.connector
371 if self.maxRetries is not None and (self.retries > self.maxRetries):
373 log.msg("Abandoning %s after %d retries." %
374 (connector, self.retries))
377 self.delay = min(self.delay * self.factor, self.maxDelay)
379 self.delay = random.normalvariate(self.delay,
380 self.delay * self.jitter)
383 log.msg("%s will retry in %d seconds" % (connector, self.delay,))
388 if self.clock is None:
389 from twisted.internet import reactor
391 self._callID = self.clock.callLater(self.delay, reconnector)
394 def stopTrying(self):
396 Put a stop to any attempt to reconnect in progress.
398 # ??? Is this function really stopFactory?
400 self._callID.cancel()
402 self.continueTrying = 0
405 self.connector.stopConnecting()
406 except error.NotConnectingError:
410 def resetDelay(self):
412 Call this method after a successful connection: it resets the delay and
415 self.delay = self.initialDelay
418 self.continueTrying = 1
421 def __getstate__(self):
423 Remove all of the state which is mutated by connection attempts and
424 failures, returning just the state which describes how reconnections
425 should be attempted. This will make the unserialized instance
426 behave just as this one did when it was first instantiated.
428 state = self.__dict__.copy()
429 for key in ['connector', 'retries', 'delay',
430 'continueTrying', '_callID', 'clock']:
437 class ServerFactory(Factory):
438 """Subclass this to indicate that your protocol.Factory is only usable for servers.
445 This is the abstract superclass of all protocols.
447 Some methods have helpful default implementations here so that they can
448 easily be shared, but otherwise the direct subclasses of this class are more
449 interesting, L{Protocol} and L{ProcessProtocol}.
454 def makeConnection(self, transport):
455 """Make a connection to a transport and a server.
457 This sets the 'transport' attribute of this Protocol, and calls the
458 connectionMade() callback.
461 self.transport = transport
462 self.connectionMade()
464 def connectionMade(self):
465 """Called when a connection is made.
467 This may be considered the initializer of the protocol, because
468 it is called when the connection is completed. For clients,
469 this is called once the connection to the server has been
470 established; for servers, this is called after an accept() call
471 stops blocking and a socket has been received. If you need to
472 send any greeting or initial message, do it here.
475 connectionDone=failure.Failure(error.ConnectionDone())
476 connectionDone.cleanFailure()
479 class Protocol(BaseProtocol):
481 This is the base class for streaming connection-oriented protocols.
483 If you are going to write a new connection-oriented protocol for Twisted,
484 start here. Any protocol implementation, either client or server, should
485 be a subclass of this class.
487 The API is quite simple. Implement L{dataReceived} to handle both
488 event-based and synchronous input; output can be sent through the
489 'transport' attribute, which is to be an instance that implements
490 L{twisted.internet.interfaces.ITransport}. Override C{connectionLost} to be
491 notified when the connection ends.
493 Some subclasses exist already to help you write common types of protocols:
494 see the L{twisted.protocols.basic} module for a few of them.
496 implements(interfaces.IProtocol, interfaces.ILoggingContext)
500 Return a prefix matching the class name, to identify log messages
501 related to this protocol instance.
503 return self.__class__.__name__
506 def dataReceived(self, data):
507 """Called whenever data is received.
509 Use this method to translate to a higher-level message. Usually, some
510 callback will be made upon the receipt of each complete protocol
513 @param data: a string of indeterminate length. Please keep in mind
514 that you will probably need to buffer some data, as partial
515 (or multiple) protocol messages may be received! I recommend
516 that unit tests for protocols call through to this method with
517 differing chunk sizes, down to one byte at a time.
520 def connectionLost(self, reason=connectionDone):
521 """Called when the connection is shut down.
523 Clear any circular references here, and any external references
524 to this Protocol. The connection has been closed.
526 @type reason: L{twisted.python.failure.Failure}
530 class ProtocolToConsumerAdapter(components.Adapter):
531 implements(interfaces.IConsumer)
533 def write(self, data):
534 self.original.dataReceived(data)
536 def registerProducer(self, producer, streaming):
539 def unregisterProducer(self):
542 components.registerAdapter(ProtocolToConsumerAdapter, interfaces.IProtocol,
543 interfaces.IConsumer)
545 class ConsumerToProtocolAdapter(components.Adapter):
546 implements(interfaces.IProtocol)
548 def dataReceived(self, data):
549 self.original.write(data)
551 def connectionLost(self, reason):
554 def makeConnection(self, transport):
557 def connectionMade(self):
560 components.registerAdapter(ConsumerToProtocolAdapter, interfaces.IConsumer,
561 interfaces.IProtocol)
563 class ProcessProtocol(BaseProtocol):
565 Base process protocol implementation which does simple dispatching for
566 stdin, stdout, and stderr file descriptors.
568 implements(interfaces.IProcessProtocol)
570 def childDataReceived(self, childFD, data):
572 self.outReceived(data)
574 self.errReceived(data)
577 def outReceived(self, data):
579 Some data was received from stdout.
583 def errReceived(self, data):
585 Some data was received from stderr.
589 def childConnectionLost(self, childFD):
591 self.inConnectionLost()
593 self.outConnectionLost()
595 self.errConnectionLost()
598 def inConnectionLost(self):
600 This will be called when stdin is closed.
604 def outConnectionLost(self):
606 This will be called when stdout is closed.
610 def errConnectionLost(self):
612 This will be called when stderr is closed.
616 def processExited(self, reason):
618 This will be called when the subprocess exits.
620 @type reason: L{twisted.python.failure.Failure}
624 def processEnded(self, reason):
626 Called when the child process exits and all file descriptors
627 associated with it have been closed.
629 @type reason: L{twisted.python.failure.Failure}
634 class AbstractDatagramProtocol:
636 Abstract protocol for datagram-oriented transports, e.g. IP, ICMP, ARP, UDP.
643 def __getstate__(self):
644 d = self.__dict__.copy()
645 d['transport'] = None
649 """Make sure startProtocol is called.
651 This will be called by makeConnection(), users should not call it.
653 if not self.numPorts:
655 log.msg("Starting protocol %s" % self)
657 self.numPorts = self.numPorts + 1
660 """Make sure stopProtocol is called.
662 This will be called by the port, users should not call it.
664 assert self.numPorts > 0
665 self.numPorts = self.numPorts - 1
666 self.transport = None
667 if not self.numPorts:
669 log.msg("Stopping protocol %s" % self)
672 def startProtocol(self):
673 """Called when a transport is connected to this protocol.
675 Will only be called once, even if multiple ports are connected.
678 def stopProtocol(self):
679 """Called when the transport is disconnected.
681 Will only be called once, after all ports are disconnected.
684 def makeConnection(self, transport):
685 """Make a connection to a transport and a server.
687 This sets the 'transport' attribute of this DatagramProtocol, and calls the
690 assert self.transport == None
691 self.transport = transport
694 def datagramReceived(self, datagram, addr):
695 """Called when a datagram is received.
697 @param datagram: the string received from the transport.
698 @param addr: tuple of source of datagram.
702 class DatagramProtocol(AbstractDatagramProtocol):
704 Protocol for datagram-oriented transport, e.g. UDP.
706 @type transport: C{NoneType} or
707 L{IUDPTransport<twisted.internet.interfaces.IUDPTransport>} provider
708 @ivar transport: The transport with which this protocol is associated,
709 if it is associated with one.
711 implements(interfaces.ILoggingContext)
715 Return a prefix matching the class name, to identify log messages
716 related to this protocol instance.
718 return self.__class__.__name__
721 def connectionRefused(self):
722 """Called due to error from write in connected mode.
724 Note this is a result of ICMP message generated by *previous*
729 class ConnectedDatagramProtocol(DatagramProtocol):
730 """Protocol for connected datagram-oriented transport.
732 No longer necessary for UDP.
735 def datagramReceived(self, datagram):
736 """Called when a datagram is received.
738 @param datagram: the string received from the transport.
741 def connectionFailed(self, failure):
742 """Called if connecting failed.
744 Usually this will be due to a DNS lookup failure.
750 """A wrapper around a file-like object to make it behave as a Transport.
752 This doesn't actually stream the file to the attached protocol,
753 and is thus useful mainly as a utility for debugging protocols.
756 implements(interfaces.ITransport)
761 streamingProducer = 0
763 def __init__(self, file):
766 def write(self, data):
768 self.file.write(data)
770 self.handleException()
771 # self._checkProducer()
773 def _checkProducer(self):
774 # Cheating; this is called at "idle" times to allow producers to be
775 # found and dealt with
777 self.producer.resumeProducing()
779 def registerProducer(self, producer, streaming):
780 """From abstract.FileDescriptor
782 self.producer = producer
783 self.streamingProducer = streaming
785 producer.resumeProducing()
787 def unregisterProducer(self):
790 def stopConsuming(self):
791 self.unregisterProducer()
792 self.loseConnection()
794 def writeSequence(self, iovec):
795 self.write("".join(iovec))
797 def loseConnection(self):
801 except (IOError, OSError):
802 self.handleException()
805 # XXX: According to ITransport, this should return an IAddress!
806 return 'file', 'file'
809 # XXX: According to ITransport, this should return an IAddress!
812 def handleException(self):
815 def resumeProducing(self):
816 # Never sends data anyways
819 def pauseProducing(self):
820 # Never sends data anyways
823 def stopProducing(self):
824 self.loseConnection()
827 __all__ = ["Factory", "ClientFactory", "ReconnectingClientFactory", "connectionDone",
828 "Protocol", "ProcessProtocol", "FileWrapper", "ServerFactory",
829 "AbstractDatagramProtocol", "DatagramProtocol", "ConnectedDatagramProtocol",