Protocols/Transports API

This part of the pulsar API is about classes responsible for implementing the Protocol/Transport paradigm. They are based on asyncio.Protocol and asyncio.DatagramProtocol classes.

Protocols

PulsarProtocol

class pulsar.async.protocols.PulsarProtocol(loop, session=1, producer=None, logger=None, **kw)[source]

A mixin class for both Protocol and DatagramProtocol.

A PulsarProtocol is an EventHandler which has two one time events:

  • connection_made
  • connection_lost
abort()[source]

Abort by aborting the transport

address

The address of the transport.

close()[source]

Close by closing the transport

Return the connection_lost event which can be used to wait for complete transport closure.

closed

True if the transport is closed.

connection_lost(exc=None)[source]

Fires the connection_lost event.

connection_made(transport)[source]

Sets the transport, fire the connection_made event and adds a timeout for idle connections.

eof_received()[source]

The socket was closed from the remote end

producer

The producer of this Protocol.

session

Connection session number.

Passed during initialisation by the producer. Usually an integer representing the number of separate connections the producer has processed at the time it created this Protocol.

sock

The socket of transport.

transport

The transport for this protocol.

Available once the connection_made() is called.

Protocol

class pulsar.async.protocols.Protocol(loop, session=1, producer=None, logger=None, **kw)[source]

An asyncio.Protocol with events

write(data)[source]

Write data into the wire.

Returns an empty tuple or a Future if this protocol has paused writing.

Connection

class pulsar.async.protocols.Connection(consumer_factory=None, timeout=None, **kw)[source]

A FlowControl to handle multiple TCP requests/responses.

It is a class which acts as bridge between a transport and a ProtocolConsumer. It routes data arriving from the transport to the current_consumer().

_consumer_factory

A factory of ProtocolConsumer.

_processed

number of separate requests processed.

current_consumer()[source]

The ProtocolConsumer currently handling incoming data.

This instance will receive data when this connection get data from the transport via the data_received() method.

If no consumer is available, build a new one and return it.

data_received(data)[source]

Delegates handling of data to the current_consumer().

Once done set a timeout for idle connections when a timeout is a positive number (of seconds).

upgrade(consumer_factory)[source]

Upgrade the _consumer_factory() callable.

This method can be used when the protocol specification changes during a response (an example is a WebSocket request/response, or HTTP tunneling).

This method adds a post_request callback to the current_consumer() to build a new consumer with the new _consumer_factory().

Parameters:consumer_factory – the new consumer factory (a callable accepting no parameters)
Returns:None.

Protocol Consumer

class pulsar.async.protocols.ProtocolConsumer(loop=None, one_time_events=None, many_times_events=None)[source]

The consumer of data for a server or client Connection.

It is responsible for receiving incoming data from an end point via the Connection.data_received() method, decoding (parsing) and, possibly, writing back to the client or server via the transport attribute.

Note

For server consumers, data_received() is the only method to implement. For client consumers, start_request() should also be implemented.

A ProtocolConsumer is a subclass of EventHandler and it has two default one time events:

  • pre_request fired when the request is received (for servers) or just before is sent (for clients). This occurs just before the start_request() method.
  • post_request fired when the request is done. The on_finished attribute is a shortcut for the post_request OneTime event and therefore can be used to wait for the request to have received a full response (clients).

In addition, it has two many times events:

  • data_received fired when new data is received from the transport but not yet processed (before the data_received() method is invoked)
  • data_processed fired just after data has been consumed (after the data_received() method)

Note

A useful example on how to use the data_received event is the wsgi proxy server.

abort_request()[source]

Abort the request.

This method can be called during the pre-request stage

connection

The Connection of this consumer.

connection_lost(exc)[source]

Called by the connection when the transport is closed.

By default it calls the finished() method. It can be overwritten to handle the potential exception exc.

connection_made(connection)[source]

Called by a Connection when it starts using this consumer.

By default it does nothing.

data_received(data)[source]

Called when some data is received.

This method must be implemented by subclasses for both server and client consumers.

The argument is a bytes object.

finished(*arg, **kw)[source]

Fire the post_request event if it wasn’t already fired.

on_finished

Event fired once a full response to a request is received. It is the post_request one time event.

producer

The Producer of this consumer.

request

The request.

Used for clients only and available only after the start() method is invoked.

start(request=None)[source]

Starts processing the request for this protocol consumer.

There is no need to override this method, implement start_request() instead. If either connection or transport are missing, a RuntimeError occurs.

For server side consumer, this method simply fires the pre_request event.

start_request()[source]

Starts a new request.

Invoked by the start() method to kick start the request with remote server. For server ProtocolConsumer this method is not invoked at all.

For clients this method should be implemented and it is critical method where errors caused by stale socket connections can arise. This method should not be called directly. Use start() instead. Typically one writes some data from the request into the transport. Something like this:

self.transport.write(self.request.encode())
transport

The Transport of this consumer

write(data)[source]

Delegate writing to the underlying Connection

Return an empty tuple or a Future

Producers

Producers are factory of Protocol with end-points. They are used by both servers and clients classes.

Producer

class pulsar.async.protocols.Producer(loop=None, protocol_factory=None, name=None, max_requests=None, logger=None)[source]

An Abstract EventHandler class for all producers of socket (client and servers)

build_consumer(consumer_factory)[source]

Build a consumer for a protocol.

This method can be used by protocols which handle several requests, for example the Connection class.

Parameters:consumer_factory – consumer factory to use.
create_protocol(**kw)[source]

Create a new protocol via the protocol_factory()

This method increase the count of sessions and build the protocol passing self as the producer.

requests_processed

Total number of requests processed.

sessions

Total number of protocols created by the Producer.

TCP Server

class pulsar.async.protocols.TcpServer(protocol_factory, loop, address=None, name=None, sockets=None, max_requests=None, keep_alive=None, logger=None)[source]

A Producer of server Connection for TCP servers.

_server

A Server managed by this Tcp wrapper.

Available once the start_serving() method has returned.

address

Socket address of this server.

It is obtained from the first socket getsockname method.

close()[source]

Stop serving the Server.sockets.

create_protocol()[source]

Override Producer.create_protocol().

start_serving(backlog=100, sslcontext=None)[source]

Start serving.

Parameters:
  • backlog – Number of maximum connections
  • sslcontext – optional SSLContext object.
Returns:

a Future called back when the server is serving the socket.

UDP

Classes for the (user) datagram protocol. UDP uses a simple transmission model with a minimum of protocol mechanism.

Datagram Protocol

class pulsar.async.protocols.DatagramProtocol(loop, session=1, producer=None, logger=None, **kw)[source]

An asyncio.DatagramProtocol with events`

Datagram Server

class pulsar.async.protocols.DatagramServer(protocol_factory, loop=None, address=None, name=None, sockets=None, max_requests=None, logger=None)[source]

An Producer for serving UDP sockets.

_transports

A list of DatagramTransport.

Available once the create_endpoint() method has returned.

close()[source]

Stop serving the Server.sockets and close all concurrent connections.

create_endpoint(**kw)[source]

create the server endpoint.

Returns:a Future called back when the server is serving the socket.

Protocol Mixins

FlowControl

class pulsar.async.mixins.FlowControl(low_limit=None, high_limit=None, **kw)[source]

A protocol mixin for flow control logic.

This implements the protocol methods pause_writing(), resume_writing().

pause_writing()[source]

Called by the transport when the buffer goes over the high-water mark

Successive calls to this method will fails unless resume_writing() is called first.

resume_writing(exc=None)[source]

Resume writing.

Successive calls to this method will fails unless pause_writing() is called first.

Timeout

class pulsar.async.mixins.Timeout[source]

Adds a timeout for idle connections to protocols

Clients

This section introduces classes implementing the transport/protocol paradigm for clients with several connections to a remote TcpServer.

Abstract Client

class pulsar.async.clients.AbstractClient(loop=None, protocol_factory=None, name=None, max_requests=None, logger=None)[source]

A Producer for a client connections.

connect()[source]

Abstract method for creating a connection.

create_connection(address, protocol_factory=None, **kw)[source]

Helper method for creating a connection to an address.

Abstract UDP Client

class pulsar.async.clients.AbstractUdpClient(loop=None, protocol_factory=None, name=None, max_requests=None, logger=None)[source]

A Producer for a client udp connections.

create_datagram_endpoint(protocol_factory=None, **kw)[source]

Helper method for creating a connection to an address.

create_endpoint()[source]

Abstract method for creating the endpoint

Pool

class pulsar.async.clients.Pool(creator, pool_size=10, loop=None, timeout=None, **kw)[source]

An asynchronous pool of open connections.

Open connections are either in_use or available to be used. Available connection are placed in an asyncio.Queue.

This class is not thread safe.

available

Number of available connections in the pool.

close()[source]

Close all connections

Return a Future called once all connections have closed

closed

True when this pool is closed

connect()[source]

Get a connection from the pool.

The connection is either a new one or retrieved from the available connections in the pool.

Returns:a Future resulting in the connection.
in_use

The number of connections in use.

These connections are not available until they are released back to the pool.

pool_size

The maximum number of open connections allowed.

If more connections are requested, the request is queued and a connection returned as soon as one becomes available.

Pool Connection

class pulsar.async.clients.PoolConnection(pool, connection)[source]

A wrapper for a Connection in a connection Pool.

pool

The Pool which created this PoolConnection

connection

The underlying socket connection.

close(discard=False)[source]

Close this pool connection by releasing the underlying connection back to the pool.

detach(discard=True)[source]

Remove the underlying connection from the connection pool.