From c0acac6d5a1a5e603571828309119ffb20e2aa7e Mon Sep 17 00:00:00 2001 From: Kevin Yu Date: Sun, 21 Jun 2015 10:19:30 -0700 Subject: [PATCH] Empty buffer after preparing _flush_endpoint in BufferedConsumer An asynchronous buffered consumer will duplicate messages if it is flushed while an earlier call to _flush_endpoint is being processed asynchronously. By emptying the buffer before iterating over the messages, subsequent calls to _flush_endpoint while async call(s) are already in progress will not duplicate messages. --- mixpanel/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/mixpanel/__init__.py b/mixpanel/__init__.py index fe7c9fb..0025222 100644 --- a/mixpanel/__init__.py +++ b/mixpanel/__init__.py @@ -440,6 +440,7 @@ def flush(self): def _flush_endpoint(self, endpoint): buf = self._buffers[endpoint] + self._buffers[endpoint] = [] while buf: batch = buf[:self._max_size] batch_json = '[{0}]'.format(','.join(batch)) @@ -451,4 +452,3 @@ def _flush_endpoint(self, endpoint): mp_e.endpoint = endpoint raise six.raise_from(mp_e, orig_e) buf = buf[self._max_size:] - self._buffers[endpoint] = buf