Skip to content
This repository was archived by the owner on Jun 23, 2020. It is now read-only.

Only use gevent for queueing #3

Merged
bderusha merged 1 commit into from
Dec 6, 2017
Merged

Only use gevent for queueing #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 4 additions & 15 deletions analytics/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,25 +3,18 @@
import logging
import numbers
import atexit
import os

from dateutil.tz import tzutc
from gevent import queue
from six import string_types

from analytics.utils import guess_timezone, clean
from analytics.consumer import Consumer
from analytics.version import VERSION

try:
from gevent import queue
except:
try:
import queue
except:
import Queue as queue


ID_TYPES = (numbers.Number, string_types)
JOIN_TIMEOUT_SECONDS = 1


class Client(object):
Expand All @@ -32,11 +25,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.log.info("import queue from {path}".format(
path=os.path.abspath(queue.__file__)
))

self.queue = queue.Queue(max_queue_size)
self.queue = 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
Expand Down Expand Up @@ -233,7 +222,7 @@ def flush(self):
"""Forces a flush from the internal queue to the server"""
queue = self.queue
size = queue.qsize()
queue.join()
queue.join(JOIN_TIMEOUT_SECONDS)
# Note that this message may not be precise, because of threading.
self.log.debug('successfully flushed about %s items.', size)

Expand Down
1 change: 1 addition & 0 deletions analytics/test/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -215,6 +215,7 @@ def test_flush(self):
# a race condition. We do our best to load it up though.
client.flush()
# Make sure that the client queue is empty after flushing
time.sleep(2)
self.assertTrue(client.queue.empty())

def test_overflow(self):
Expand Down
2 changes: 1 addition & 1 deletion analytics/version.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1 +1 @@
VERSION = '1.2.8'
VERSION = '1.2.10'
3 changes: 2 additions & 1 deletion setup.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,8 @@
install_requires = [
"requests>=2.7,<3.0",
"six>=1.5",
"python-dateutil>2.1"
"python-dateutil>2.1",
"gevent>=1.0.2"
]

setup(
Expand Down