Merge pull request #2 from appsembler/shadinaif/use-multiprocessing

Use JoinableQueue for compatibility with Celery

Author: shadinaif

.github/workflows/ci.yml

@@ -0,0 +1,28 @@
+name: ci
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    name: ${{ matrix.tox-env }}
+    strategy:
+      matrix:
+        include:
+          - python-version: 3.8
+            tox-env: py38
+          - python-version: 3.5
+            tox-env: py35
+    steps:
+      - uses: actions/checkout@v3
+      - name: Set up Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install dependencies
+        run: pip install tox
+      - name: Test with tox
+        run: tox -e ${{ matrix.tox-env }}

analytics/client.py

@@ -11,10 +11,8 @@
 from analytics.consumer import Consumer
 from analytics.version import VERSION
 
-try:
-    import queue
-except:
-    import Queue as queue
+from queue import Full
+from multiprocessing import JoinableQueue
 
 
 ID_TYPES = (numbers.Number, string_types)
@@ -28,7 +26,7 @@ def __init__(self, write_key=None, host=None, debug=False, max_queue_size=10000,
                  send=True, on_error=None):
         require('write_key', write_key, string_types)
 
-        self.queue = queue.Queue(max_queue_size)
+        self.queue = JoinableQueue(max_queue_size)
         self.consumer = Consumer(self.queue, write_key, host=host, on_error=on_error)
         self.write_key = write_key
         self.on_error = on_error
@@ -220,7 +218,7 @@ def _enqueue(self, msg):
             self.queue.put(msg, block=False)
             self.log.debug('enqueued %s.', msg['type'])
             return True, msg
-        except queue.Full:
+        except Full:
             self.log.warn('analytics-python queue is full')
             return False, msg
 

analytics/consumer.py

@@ -4,10 +4,8 @@
 from analytics.version import VERSION
 from analytics.request import post
 
-try:
-    from queue import Empty
-except:
-    from Queue import Empty
+from queue import Empty
+
 
 class Consumer(Thread):
     """Consumes the messages from the client's queue."""

analytics/test/consumer.py

@@ -1,24 +1,21 @@
 import unittest
 
-try:
-    from queue import Queue
-except:
-    from Queue import Queue
+from multiprocessing import JoinableQueue
 
 from analytics.consumer import Consumer
 
 
 class TestConsumer(unittest.TestCase):
 
     def test_next(self):
-        q = Queue()
+        q = JoinableQueue()
         consumer = Consumer(q, '')
         q.put(1)
         next = consumer.next()
         self.assertEqual(next, [1])
 
     def test_next_limit(self):
-        q = Queue()
+        q = JoinableQueue()
         upload_size = 50
         consumer = Consumer(q, '', upload_size)
         for i in range(10000):
@@ -27,7 +24,7 @@ def test_next_limit(self):
         self.assertEqual(next, list(range(upload_size)))
 
     def test_upload(self):
-        q = Queue()
+        q = JoinableQueue()
         consumer  = Consumer(q, 'testsecret')
         track = {
             'type': 'track',

tox.ini

@@ -0,0 +1,16 @@
+[tox]
+envlist = py35,py38
+
+[testenv]
+download = True
+
+deps =
+    python-dateutil==2.4.0
+    pytest
+
+commands =
+    pytest ./analytics/test/client.py \
+        ./analytics/test/consumer.py \
+        ./analytics/test/module.py \
+        ./analytics/test/request.py \
+        ./analytics/test/utils.py