Protocols/Transports API — pulsar 1.6.1 documentation (original) (raw)

This part of the pulsar API is about classes responsible for implementing the Protocol/Transport paradigm. They are based onasyncio.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 andDatagramProtocol.

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 thisProtocol.

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 atransport and a ProtocolConsumer. It routes data arriving from the transport to thecurrent_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 thedata_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 atimeout 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 thecurrent_consumer() to build a new consumer with the new_consumer_factory().

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 theConnection.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. Theon_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 thedata_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 thestart() 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, aRuntimeError occurs.

For server side consumer, this method simply fires the pre_requestevent.

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 requestinto 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.

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.

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.

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 unlessresume_writing() is called first.

resume_writing(exc=None)[source]¶

Resume writing.

Successive calls to this method will fails unlesspause_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 availableto 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 theavailable connections in the pool.

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 underlyingconnection back to the pool.

detach(discard=True)[source]¶

Remove the underlying connection from the connectionpool.