Merge lp:~diogobaeder/txstatsd/fix-host-lookup into lp:txstatsd

[Diogo Baeder]

Sending messages to a 'queue' (a list, actually, to make it simpler) while the host is not resolved, and flushing them as soon as the host gets resolved. I didn't use Queue.Queue here because it doesn't seem that we'll have to deal with concurrency issues with this queue, since each client instance has its own queue, but if we feel the need, this can be changed later.

Sending messages to a 'queue' (a list, actually, to make it simpler) while the host is not resolved, and flushing them as soon as the host gets resolved. I didn't use Queue.Queue here because it doesn't seem that we'll have to deal with concurrency issues with this queue, since each client instance has its own queue, but if we feel the need, this can be changed later.

[Sidnei da Silva (sidnei)]

A few comments:

1. I dislike class attributes with parameters that cannot be set through the constructor. How are you suppose to change LIMIT? By subclassing? By monkey patching?

2. Along the same lines, because you can't easily change LIMIT, the tests iterate calling write() up to 'LIMIT' times, which is useless. If LIMIT was configurable, you could set it to '1' in the tests and make them more efficient.

review: Needs Fixing

[Diogo Baeder (diogobaeder)]

Fixed. Can you check again?

Em 26-12-2012 17:21, Sidnei da Silva escreveu:
> Review: Needs Fixing
>
> A few comments:
>
> 1. I dislike class attributes with parameters that cannot be set through the constructor. How are you suppose to change LIMIT? By subclassing? By monkey patching?
>
> 2. Along the same lines, because you can't easily change LIMIT, the tests iterate calling write() up to 'LIMIT' times, which is useless. If LIMIT was configurable, you could set it to '1' in the tests and make them more efficient.

[Sidnei da Silva (sidnei)]

You still haven't changed the test to use limit=1. :)

[Diogo Baeder (diogobaeder)]

I changed it to use limit=2 instead of limit=1 because it's more
expressive to me that the limit is actually used/respected when queueing
the calls (if I use only 1, it's not clear that it respects the number
given as a limit or if it closes right after the first message sent to
the queue). This shouldn't have any noticeable impact on the test duration.

Would this be OK for you?

Em 26-12-2012 18:08, Sidnei da Silva escreveu:
> You still haven't changed the test to use limit=1. :)

[Diogo Baeder (diogobaeder)]

Changed again to use limit=2 in the default instance for tests, instead
of in specific tests, as requested.

Em 26-12-2012 18:39, Diogo Baeder escreveu:
> I changed it to use limit=2 instead of limit=1 because it's more
> expressive to me that the limit is actually used/respected when queueing
> the calls (if I use only 1, it's not clear that it respects the number
> given as a limit or if it closes right after the first message sent to
> the queue). This shouldn't have any noticeable impact on the test duration.
>
> Would this be OK for you?
>
> Em 26-12-2012 18:08, Sidnei da Silva escreveu:
>> You still haven't changed the test to use limit=1. :)
>

[Sidnei da Silva (sidnei)]

I'm fairly confident that calling transport.connect() does not do what you want. That sets the transport to 'connected udp' which is not what we want. See: http://twistedmatrix.com/documents/11.0.0/core/howto/udp.html#auto2

Instead, you should pass host, port to the transport gateway and use them in transport.write(data, (host, port)).

[Diogo Baeder (diogobaeder)]

Well, as far as I understood it, it doesn't create a "real connection",
only makes sure that all datagrams are delivered (when they are) to the
same host. As what we want is to send datagrams to the same host (as we
even resolve the host before sending the packets), this makes perfect
sense - and makes it cleaner to send messages, as we only have to pass
the data as an argument, instead of providing (host, port) for every
write() call.

Here, take a look at the paragraphs of the documentation that make it clear:
"A connected UDP socket is slightly different from a standard one - it
can only send and receive datagrams to/from a single address, but this
does not in any way imply a connection"
"Unlike a regular UDP protocol, we do not need to specify where to send
datagrams and are not told where they came from since they can only come
from the address to which the socket is 'connected'."

Em 27-12-2012 13:59, Sidnei da Silva escreveu:
> I'm fairly confident that calling transport.connect() does not do what you want. That sets the transport to 'connected udp' which is not what we want. See: http://twistedmatrix.com/documents/11.0.0/core/howto/udp.html#auto2
>
> Instead, you should pass host, port to the transport gateway and use them in transport.write(data, (host, port)).
>
>

[Sidnei da Silva (sidnei)]

Yes, but we tried that before and it didn't work as intended. We don't care about making sure that the datagrams are delivered so it's only adding overhead to it.

[Diogo Baeder (diogobaeder)]

Ah, ok, good to know that, I'll avoid trying to fix the host in the
future, then. Now I changed it to use explicit (host, port) on every call.

Diff updated, can you take a look again?

Em 27-12-2012 15:24, Sidnei da Silva escreveu:
> Yes, but we tried that before and it didn't work as intended. We don't care about making sure that the datagrams are delivered so it's only adding overhead to it.

[Sidnei da Silva (sidnei)]

review: Approve

[Ubuntu One Server Tarmac Bot (ubuntuone-server-tarmac)]

The attempt to merge lp:~diogobaeder/txstatsd/fix-host-lookup into lp:txstatsd failed. Below is the output from the failed tests.

txstatsd.tests.metrics.test_distinct
  TestDistinct
    test_all ... [OK]
  TestDistinctMetricReporter
    test_reports ... [OK]
  TestHash
    test_chi_square ... [SKIPPED]
    test_hash_chars ... [OK]
  TestPlugin
    test_factory ... [OK]
  TestZeros
    test_zeros ... [OK]
txstatsd.tests.metrics.test_histogrammetric
  TestHistogramReporterMetric
    test_histogram_histogram ... [OK]
    test_histogram_of_numbers_1_through_10000 ... [OK]
    test_histogram_with_zero_recorded_values ... [OK]
txstatsd.tests.metrics.test_metermetric
  TestDeriveMetricReporter
    test_fastpoll ... [OK]
    test_interface ... [OK]
txstatsd.tests.metrics.test_sli
  TestConditions
    test_above ... [OK]
    test_above_linear ... [OK]
    test_below ... [OK]
    test_below_linear ... [OK]
    test_between ... [OK]
  TestFactory
    test_configure ... [OK]
    test_configure_linear ... [OK]
  TestMetric
    test_clear ... [OK]
    test_count_all ... [OK]
    test_count_error ... [OK]
    test_count_threshold ... [OK]
    test_reports ... [OK]
  TestMetricLinear
    test_count_threshold ... [OK]
  TestParsing
    test_parse ... [OK]
txstatsd.tests.metrics.test_timermetric
  TestBlankTimerMetric
    test_count ... [OK]
    test_max ... [OK]
    test_mean ... [OK]
    test_min ... [OK]
    test_no_values ... [OK]
    test_percentiles ... [OK]
    test_rate ... [OK]
    test_std_dev ... [OK]...

[lp:~diogobaeder/txstatsd/fix-host-lookup]

117. By Diogo Baeder on 2012-12-27

Fixing issues with connection stablished after the host resolves and the gateway is instantiated

=== modified file 'txstatsd/protocol.py'
--- txstatsd/protocol.py	2012-12-18 19:53:43 +0000
+++ txstatsd/protocol.py	2012-12-27 20:51:21 +0000
@@ -26,6 +26,9 @@
 from twisted.python import log
 
 
+__all__ = ('StatsDClientProtocol', 'TwistedStatsDClient')
+
+
 class StatsDClientProtocol(DatagramProtocol):
     """A Twisted-based implementation of the StatsD client protocol.
 
@@ -44,18 +47,82 @@
         self.client.disconnect()
 
 
+class DataQueue(object):
+    """Manages the queue of sent data, so that it can be really sent later when
+    the host is resolved."""
+
+    def __init__(self, limit=1000):
+        self._limit = limit
+        self._queue = []
+
+    def write(self, data, callback):
+        """Queue the given data, so that it's sent later.
+
+        @param data: The data to be queued.
+        @param callback: The callback to use when the data is flushed.
+        """
+        if len(self._queue) < self._limit:
+            self._queue.append((data, callback))
+
+    def flush(self):
+        """Flush the queue, returning its items."""
+        items = self._queue
+        self._queue = []
+        return items
+
+
+class TransportGateway(object):
+    """Responsible for sending datagrams to the actual transport."""
+
+    def __init__(self, transport, reactor, host, port):
+        """
+        @param transport: DatagramProtocol().transport .
+        @param reactor: The Twisted reactor in use.
+        """
+        self.transport = transport
+        self.reactor = reactor
+        self.host = host
+        self.port = port
+
+    def write(self, data, callback):
+        """Send the metric to the StatsD server.
+
+        @param data: The data to be sent.
+        @param callback: The callback to which the result should be sent.
+            B{Note}: The C{callback} will be called in the C{reactor}
+            thread, and not in the thread of the original caller.
+        """
+        self.reactor.callFromThread(self._write, data, callback)
+
+    def _write(self, data, callback):
+        """Send the metric to the StatsD server.
+
+        @param data: The data to be sent.
+        @param callback: The callback to which the result should be sent.
+        @raise twisted.internet.error.MessageLengthError: If the size of data
+            is too large.
+        """
+        try:
+            bytes_sent = self.transport.write(data, (self.host, self.port))
+            if callback is not None:
+                callback(bytes_sent)
+        except (OverflowError, TypeError, socket.error, socket.gaierror):
+            if callback is not None:
+                callback(None)
+
+
 class TwistedStatsDClient(object):
 
     def __init__(self, host, port, connect_callback=None,
-                 disconnect_callback=None, resolver_errback=None):
-        """
-        Build a connection that reports to the endpoint (on C{host} and
-        C{port}) using UDP.
+                 disconnect_callback=None):
+        """Avoid using this initializer directly; Instead, use the create()
+        static method, otherwise the messages won't be really delivered.
+
+        If you still need to use this directly and want to resolve the host
+        yourself, remember to call host_resolved() as soon as it's resolved.
 
         @param host: The StatsD server host.
         @param port: The StatsD server port.
-        @param resolver_errback: Deprecated parameter, unused.
-            Please avoid using it.
         @param connect_callback: The callback to invoke on connection.
         @param disconnect_callback: The callback to invoke on disconnection.
         """
@@ -64,12 +131,13 @@
         self.reactor = reactor
 
         self.host = host
-
         self.port = port
         self.connect_callback = connect_callback
         self.disconnect_callback = disconnect_callback
+        self.data_queue = DataQueue()
 
         self.transport = None
+        self.transport_gateway = None
 
     def __str__(self):
         return "%s:%d" % (self.host, self.port)
@@ -77,7 +145,9 @@
     @staticmethod
     def create(host, port, connect_callback=None, disconnect_callback=None,
                resolver_errback=None):
-        """Resolve the host and return a Deferred for the instance.
+        """Create an instance that resolves the host to an IP asynchronously.
+
+        Will queue all messages while the host is not yet resolved.
 
         Build a connection that reports to the endpoint (on C{host} and
         C{port}) using UDP.
@@ -90,28 +160,26 @@
         @param disconnect_callback: The callback to invoke on disconnection."""
         from twisted.internet import reactor
 
-        deferred_instance = Deferred()
-
-        def create_instance(ip):
-            instance = TwistedStatsDClient(
-                host=ip, port=port, connect_callback=connect_callback,
-                disconnect_callback=disconnect_callback)
-            deferred_instance.callback(instance)
+        instance = TwistedStatsDClient(
+            host=host, port=port, connect_callback=connect_callback,
+            disconnect_callback=disconnect_callback)
 
         if resolver_errback is None:
             resolver_errback = log.err
 
-        resolver = reactor.resolve(host)
-        resolver.addCallbacks(create_instance, resolver_errback)
+        instance.resolve_later = reactor.resolve(host)
+        instance.resolve_later.addCallbacks(instance.host_resolved,
+                                            resolver_errback)
 
-        return deferred_instance
+        return instance
 
     def connect(self, transport=None):
         """Connect to the StatsD server."""
         if transport is not None:
             self.transport = transport
-            if self.connect_callback is not None:
-                self.connect_callback()
+            if self.transport_gateway is not None:
+                self.transport_gateway.transport = transport
+        self._flush_items()
 
     def disconnect(self):
         """Disconnect from the StatsD server."""
@@ -127,21 +195,24 @@
             B{Note}: The C{callback} will be called in the C{reactor}
             thread, and not in the thread of the original caller.
         """
-        self.reactor.callFromThread(self._write, data, callback)
-
-    def _write(self, data, callback):
-        """Send the metric to the StatsD server.
-
-        @param data: The data to be sent.
-        @param callback: The callback to which the result should be sent.
-        @raise twisted.internet.error.MessageLengthError: If the size of data
-            is too large.
-        """
-        if self.host is not None and self.transport is not None:
-            try:
-                bytes_sent = self.transport.write(data, (self.host, self.port))
-                if callback is not None:
-                    callback(bytes_sent)
-            except (OverflowError, TypeError, socket.error, socket.gaierror):
-                if callback is not None:
-                    callback(None)
+        if self.transport_gateway is not None and self.transport is not None:
+            return self.transport_gateway.write(data, callback)
+        return self.data_queue.write(data, callback)
+
+    def host_resolved(self, ip):
+        """Callback used when the host is resolved to an IP address."""
+        self.host = ip
+        self.transport_gateway = TransportGateway(self.transport, self.reactor,
+                                                  self.host, self.port)
+
+        if self.connect_callback is not None:
+            self.connect_callback()
+
+        self._flush_items()
+
+    def _flush_items(self):
+        """Flush all items (data, callback) from the DataQueue to the
+        TransportGateway."""
+        for item in self.data_queue.flush():
+            data, callback = item
+            self.write(data, callback)

=== modified file 'txstatsd/server/router.py'
--- txstatsd/server/router.py	2012-06-28 17:29:26 +0000
+++ txstatsd/server/router.py	2012-12-27 20:51:21 +0000
@@ -264,7 +264,7 @@
         d = defer.Deferred()
         self.ready.addCallback(lambda _: d)
 
-        client = TwistedStatsDClient(
+        client = TwistedStatsDClient.create(
             host, port, connect_callback=lambda: d.callback(None))
         protocol = StatsDClientProtocol(client)
 

=== modified file 'txstatsd/tests/test_client.py'
--- txstatsd/tests/test_client.py	2012-12-18 19:53:43 +0000
+++ txstatsd/tests/test_client.py	2012-12-27 20:51:21 +0000
@@ -21,6 +21,8 @@
 """Tests for the various client classes."""
 
 import sys
+
+from mocker import Mocker, expect, ANY
 from twisted.internet import reactor
 from twisted.internet.defer import inlineCallbacks, Deferred
 from twisted.python import log
@@ -32,7 +34,9 @@
 from txstatsd.metrics.metric import Metric
 from txstatsd.client import (
     StatsDClientProtocol, TwistedStatsDClient, UdpStatsDClient,
-    ConsistentHashingClient)
+    ConsistentHashingClient
+)
+from txstatsd.protocol import DataQueue, TransportGateway
 
 
 class FakeClient(object):
@@ -63,16 +67,21 @@
         super(TestClient, self).setUp()
         self.client = None
         self.exception = None
+        self.mocker = Mocker()
 
     def tearDown(self):
         if self.client:
             self.client.transport.stopListening()
         super(TestClient, self).tearDown()
 
+    def build_protocol(self):
+        protocol = StatsDClientProtocol(self.client)
+        reactor.listenUDP(0, protocol)
+
     def test_twistedstatsd_write(self):
         self.client = TwistedStatsDClient('127.0.0.1', 8000)
-        protocol = StatsDClientProtocol(self.client)
-        reactor.listenUDP(0, protocol)
+        self.build_protocol()
+        self.client.host_resolved('127.0.0.1')
 
         def ensure_bytes_sent(bytes_sent):
             self.assertEqual(bytes_sent, len('message'))
@@ -87,10 +96,10 @@
 
     @inlineCallbacks
     def test_twistedstatsd_write_with_host_resolved(self):
-        self.client = yield TwistedStatsDClient.create(
+        self.client = TwistedStatsDClient.create(
             'localhost', 8000)
-        protocol = StatsDClientProtocol(self.client)
-        reactor.listenUDP(0, protocol)
+        self.build_protocol()
+        yield self.client.resolve_later
 
         def ensure_bytes_sent(bytes_sent):
             self.assertEqual(bytes_sent, len('message'))
@@ -106,36 +115,38 @@
 
     @inlineCallbacks
     def test_twistedstatsd_with_malformed_address_and_errback(self):
-        def ensure_exception_raised(exception):
-            self.assertTrue(exception.startswith("DNS lookup failed"))
+        exceptions_captured = []
 
         def capture_exception_raised(failure):
             exception = failure.getErrorMessage()
-            self.deferred_instance.callback(exception)
+            self.assertTrue(exception.startswith("DNS lookup failed"))
+            exceptions_captured.append(exception)
 
-        self.deferred_instance = TwistedStatsDClient.create(
+        self.client = TwistedStatsDClient.create(
             '256.0.0.0', 1,
             resolver_errback=capture_exception_raised)
+        self.build_protocol()
+        yield self.client.resolve_later
 
-        self.deferred_instance.addCallback(ensure_exception_raised)
-        yield self.deferred_instance
+        self.assertEqual(len(exceptions_captured), 1)
 
     @inlineCallbacks
     def test_twistedstatsd_with_malformed_address_and_no_errback(self):
-        def ensure_exception_raised(exception):
-            self.assertTrue(exception.startswith("DNS lookup failed"))
+        exceptions_captured = []
 
         def capture_exception_raised(failure):
             exception = failure.getErrorMessage()
-            self.deferred_instance.callback(exception)
+            self.assertTrue(exception.startswith("DNS lookup failed"))
+            exceptions_captured.append(exception)
 
         self.patch(log, "err", capture_exception_raised)
 
-        self.deferred_instance = TwistedStatsDClient.create(
+        self.client = TwistedStatsDClient.create(
             '256.0.0.0', 1)
+        self.build_protocol()
+        yield self.client.resolve_later
 
-        self.deferred_instance.addCallback(ensure_exception_raised)
-        yield self.deferred_instance
+        self.assertEqual(len(exceptions_captured), 1)
 
     def test_udpstatsd_wellformed_address(self):
         client = UdpStatsDClient('localhost', 8000)
@@ -181,6 +192,187 @@
             if 'twisted' in mod:
                 self.assertTrue(sys.modules[mod] is None)
 
+    def test_starts_with_data_queue(self):
+        """The client starts with a DataQueue."""
+        self.client = TwistedStatsDClient('127.0.0.1', 8000)
+        self.build_protocol()
+
+        self.assertIsInstance(self.client.data_queue, DataQueue)
+
+    def test_starts_without_transport_gateway(self):
+        """The client starts without a TransportGateway."""
+        self.client = TwistedStatsDClient('127.0.0.1', 8000)
+        self.build_protocol()
+
+        self.assertTrue(self.client.transport_gateway is None)
+
+    def test_passes_transport_to_gateway(self):
+        """The client passes the transport to the gateway as soon as the client
+        is connected."""
+        self.client = TwistedStatsDClient('127.0.0.1', 8000)
+        self.build_protocol()
+        self.client.host_resolved('127.0.0.1')
+
+        self.assertEqual(self.client.transport_gateway.transport,
+                         self.client.transport)
+
+    def test_passes_reactor_to_gateway(self):
+        """The client passes the reactor to the gateway as soon as the client
+        is connected."""
+        self.client = TwistedStatsDClient('127.0.0.1', 8000)
+        self.build_protocol()
+        self.client.host_resolved('127.0.0.1')
+
+        self.assertEqual(self.client.transport_gateway.reactor,
+                         self.client.reactor)
+
+    def test_sets_ip_when_host_resolves(self):
+        """As soon as the host is resolved, set the IP as the host."""
+        self.client = TwistedStatsDClient('localhost', 8000)
+        self.build_protocol()
+        self.assertEqual(self.client.host, 'localhost')
+
+        self.client.host_resolved('127.0.0.1')
+        self.assertEqual(self.client.host, '127.0.0.1')
+
+    def test_sets_transport_gateway_when_host_resolves(self):
+        """As soon as the host is resolved, set the transport gateway."""
+        self.client = TwistedStatsDClient('localhost', 8000)
+        self.build_protocol()
+
+        self.client.transport_gateway = None
+
+        self.client.host_resolved('127.0.0.1')
+        self.assertIsInstance(self.client.transport_gateway, TransportGateway)
+
+    def test_calls_connect_callback_when_host_resolves(self):
+        """As soon as the host is resolved, call back the connect_callback."""
+        self.client = TwistedStatsDClient('localhost', 8000)
+        self.build_protocol()
+
+        self.client.connect_callback = self.mocker.mock()
+        expect(self.client.connect_callback())
+
+        with self.mocker:
+            self.client.host_resolved('127.0.0.1')
+
+    def test_sends_messages_to_gateway_after_host_resolves(self):
+        """After the host is resolved, send messages to the
+        TransportGateway."""
+        self.client = TwistedStatsDClient('localhost', 8000)
+        self.build_protocol()
+        self.client.host_resolved('127.0.0.1')
+
+        message = 'some data'
+        bytes_sent = len(message)
+        self.client.data_queue = self.mocker.mock(spec=DataQueue)  # not called
+        self.client.transport_gateway = self.mocker.mock(spec=TransportGateway)
+        callback = self.mocker.mock()
+        expect(self.client.transport_gateway.write(message, callback)).result(
+            bytes_sent)
+
+        with self.mocker:
+            self.assertEqual(self.client.write(message, callback), bytes_sent)
+
+    def test_sends_messages_to_queue_before_host_resolves(self):
+        """Before the host is resolved, send messages to the DataQueue."""
+        self.client = TwistedStatsDClient('localhost', 8000)
+        self.build_protocol()
+
+        message = 'some data'
+        bytes_sent = len(message)
+        self.client.data_queue = self.mocker.mock(spec=DataQueue)
+        callback = self.mocker.mock()
+        expect(self.client.data_queue.write(message, callback)).result(None)
+
+        with self.mocker:
+            self.assertEqual(self.client.write(message, callback), None)
+
+    def test_flushes_queued_messages_to_the_gateway_when_host_resolves(self):
+        """As soon as the host is resolved, flush all messages to the
+        TransportGateway."""
+        self.client = TwistedStatsDClient('localhost', 8000)
+        self.build_protocol()
+
+        self.client.data_queue.write('data 1', 'callback 1')
+        self.client.data_queue.write('data 2', 'callback 2')
+        self.client.data_queue.write('data 3', 'callback 3')
+
+        mock_gateway_write = self.mocker.mock()
+        self.patch(TransportGateway, 'write', mock_gateway_write)
+        expect(mock_gateway_write('data 1', 'callback 1'))
+        expect(mock_gateway_write('data 2', 'callback 2'))
+        expect(mock_gateway_write('data 3', 'callback 3'))
+
+        with self.mocker:
+            self.client.host_resolved('127.0.0.1')
+
+    def test_sets_client_transport_when_connected(self):
+        """Set the transport as an attribute of the client."""
+        self.client = TwistedStatsDClient('localhost', 8000)
+        transport = DummyTransport()
+        self.client.connect(transport)
+
+        self.assertEqual(self.client.transport, transport)
+
+    def test_sets_gateway_transport_when_connected(self):
+        """Set the transport as an attribute of the TransportGateway."""
+        self.client = TwistedStatsDClient('localhost', 8000)
+        self.client.host_resolved('127.0.0.1')
+        transport = DummyTransport()
+        self.client.connect(transport)
+
+        self.assertEqual(self.client.transport_gateway.transport, transport)
+
+
+class DataQueueTest(TestCase):
+    """Tests for the DataQueue class."""
+
+    def setUp(self):
+        super(DataQueueTest, self).setUp()
+        self.queue = DataQueue(limit=2)
+
+    def test_queues_messages_and_callbacks(self):
+        """All messages are queued with their respective callbacks."""
+        self.queue.write(data=1, callback='1')
+        self.queue.write(data=2, callback='2')
+
+        self.assertEqual(self.queue.flush(), [
+            (1, '1'),
+            (2, '2'),
+        ])
+
+    def test_flushes_the_queue(self):
+        """All messages are queued with their respective callbacks."""
+        self.queue.write(data=1, callback='1')
+        self.queue.write(data=2, callback='2')
+
+        self.queue.flush()
+        self.assertEqual(self.queue.flush(), [])
+
+    def test_limits_number_of_messages(self):
+        """Cannot save more messages than the defined limit."""
+        self.queue.write('saved data', 'saved callback')
+        self.queue.write('saved data', 'saved callback')
+        self.queue.write('discarded data', 'discarded message')
+
+        self.assertEqual(len(self.queue.flush()), 2)
+
+    def test_discards_messages_after_limit(self):
+        """Cannot save more messages than the defined limit."""
+        self.queue.write('saved data', 'saved callback')
+        self.queue.write('saved data', 'saved callback')
+        self.queue.write('discarded data', 'discarded message')
+
+        self.assertEqual(set(self.queue.flush()),
+                         set([('saved data', 'saved callback')]))
+
+    def test_makes_limit_optional(self):
+        """Use the default limit when not given."""
+        queue = DataQueue()
+
+        self.assertTrue(queue._limit > 0)
+
 
 class TestConsistentHashingClient(TestCase):
 
@@ -251,3 +443,8 @@
         client.disconnect()
         self.assertTrue(clients[0].disconnect_called)
         self.assertTrue(clients[1].disconnect_called)
+
+
+class DummyTransport(object):
+    def stopListening(self):
+        pass