Using DeferredQueue for inter-task communication in Twisted

Question

I have a Client that currently does the following:

1. connects
2. collects some data locally
3. sends that data to a server
4. repeats
5. if disconnected, reconnects and continues the above (not shown)

Like this:

def do_send(self):
    def get_data():
        # do something
        return data

    def send_data(data)
        self.sendMessage(data)

    return deferToThread(get_data).addCallback(send_data)

def connectionMade(self):
    WebSocketClientProtocol.connectionMade(self)
    self.sender = task.LoopingCall(self.do_send)
    self.sender.start(60)

However, when disconnected, I would like the data collection to continue, probably queuing and writing to file at a certain limit. I have reviewed the DeferredQueue object which seems like what I need, but I can't seem to crack it.

In pseudo-code, it would go something like this:

queue = DeferredQueue

# in a separate class from the client protocol
def start_data_collection():
    self.collecter = task.LoopingCall(self.get_data)
    self.sender.start(60)

def get_data()
    # do something
    queue.put(data)

Then have the client protocol check the queue, which is where I get lost. Is DeferredQueue what I need, or is there a better way?

Answer 1

A list would work just as well. You'll presumably get lost in the same place - how do you have the client protocol check the list?

Either way, here's one answer:

queued = []

...

connecting = endpoint.connect(factory)
def connected(protocol):
    if queued:
        sending = protocol.sendMessage(queued.pop(0))
        sending.addCallback(sendNextMessage, protocol)
        sending.addErrback(reconnect)
connecting.addCallback(connected)

The idea here is that at some point an event happens: your connection is established. This example represents that event as the connecting Deferred. When the event happens, connected is called. This example pops the first item from the queue (a list) and sends it. It waits for the send to be acknowledged and then sends the next message. It also implies some logic about handling errors by reconnecting.

Your code could look different. You could use the Protocol.connectionMade callback to represent the connection event instead. The core idea is the same - define callbacks to handle certain events when they happen. Whether you use an endpoint's connect Deferred or a protocol's connectionMade doesn't really matter.