From 3bf0c051c46c1ba2e3f6ae83bd4d2dc09c4d6f19 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Fri, 11 Nov 2016 22:35:26 +0100
Subject: [PATCH 01/26] implement stdlib queue API

---
 persistent_queue/__init__.py | 128 +++++++++++++++++++++++++++++++++++
 1 file changed, 128 insertions(+)

diff --git a/persistent_queue/__init__.py b/persistent_queue/__init__.py
index d37a7ed..59f02dd 100644
--- a/persistent_queue/__init__.py
+++ b/persistent_queue/__init__.py
@@ -31,6 +31,9 @@ def __init__(self, filename, path='.', dumps=pickle.dumps,
         self.pop_lock = threading.RLock()
         self.pushed_event = threading.Event()
 
+        self.all_tasks_done = threading.Condition()
+        self.unfinished_tasks = 0
+
         self.file.seek(0, 0)
         self.length = struct.unpack(HEADER_STRUCT[0], self.file.read(4))[0]
 
@@ -281,6 +284,7 @@ def write_data(item):
                 write_data(i)
 
             self._update_length(self.count() + len(items))
+            self.unfinished_tasks += 1
 
         self.pushed_event.set()
         _LOGGER.debug("Done pushing data")
@@ -288,3 +292,127 @@ def write_data(item):
     def __len__(self):
         """Get size of queue."""
         return self.count()
+
+    def qsize(self):
+        """
+        Provides compatibility with stdlib Queue objects.
+
+        Return the approximate size of the queue. Note, qsize() > 0 doesn’t
+        guarantee that a subsequent get() will not block, nor will qsize() <
+        maxsize guarantee that put() will not block.
+        """
+        return self.count()
+
+    def empty(self):
+        """
+        Provides compatibility with stdlib Queue objects.
+
+        Return True if the queue is empty, False otherwise. If empty() returns
+        True it doesn’t guarantee that a subsequent call to put() will not
+        block. Similarly, if empty() returns False it doesn’t guarantee that a
+        subsequent call to get() will not block.
+        """
+        return self.count() == 0
+
+    def full(self):
+        """
+        Provides compatibility with stdlib Queue objects.
+
+        Return True if the queue is full, False otherwise. If full() returns
+        True it doesn’t guarantee that a subsequent call to get() will not
+        block. Similarly, if full() returns False it doesn’t guarantee that a
+        subsequent call to put() will not block.
+        """
+        return False
+
+    def put(self, item, block=True, timeout=None):
+        """
+        Provides compatibility with stdlib Queue objects.
+        The arguments block and timeout are ignored.
+
+        Put item into the queue. If optional args block is true and timeout is
+        None (the default), block if necessary until a free slot is available.
+        If timeout is a positive number, it blocks at most timeout seconds and
+        raises the Full exception if no free slot was available within that
+        time. Otherwise (block is false), put an item on the queue if a free
+        slot is immediately available, else raise the Full exception (timeout
+        is ignored in that case).
+        """
+        self.push(item)
+
+    def put_nowait(self, item):
+        """
+        Provides compatibility with stdlib Queue objects.
+
+        Equivalent to put(item, False).
+        """
+        self.put(item, block=False)
+
+    def get(self, block=True, timeout=None):
+        """
+        Provides compatibility with stdlib Queue objects.
+        The argument timeout is ignored.
+
+        Remove and return an item from the queue. If optional args block is
+        true and timeout is None (the default), block if necessary until an
+        item is available. If timeout is a positive number, it blocks at most
+        timeout seconds and raises the Empty exception if no item was available
+        within that time. Otherwise (block is false), return an item if one is
+        immediately available, else raise the Empty exception (timeout is
+        ignored in that case).
+        """
+        item = self.pop(blocking=block)
+        if item is None:
+            raise queue.Empty
+        return item
+
+    def get_nowait(self):
+        """
+        Provides compatibility with stdlib Queue objects.
+
+        Equivalent to get(False).
+
+        Two methods are offered to support tracking whether enqueued tasks
+        have been fully processed by daemon consumer threads.
+        """
+        return self.get(block=False)
+
+    def task_done(self):
+        """
+        Provides compatibility with stdlib Queue objects.
+        Taken from the Python 3.5 stdlib: lib/python3.5/queue.py
+
+        Indicate that a formerly enqueued task is complete. Used by queue
+        consumer threads. For each get() used to fetch a task, a subsequent
+        call to task_done() tells the queue that the processing on the task is
+        complete.
+
+        If a join() is currently blocking, it will resume when all items have
+        been processed (meaning that a task_done() call was received for every
+        item that had been put() into the queue).
+
+        Raises a ValueError if called more times than there were items placed in the queue.
+        """
+        with self.all_tasks_done:
+            unfinished = self.unfinished_tasks - 1
+            if unfinished <= 0:
+                if unfinished < 0:
+                    raise ValueError('task_done() called too many times')
+                self.all_tasks_done.notify_all()
+            self.unfinished_tasks = unfinished
+
+    def join(self):
+        """
+        Provides compatibility with stdlib Queue objects.
+        Taken from the Python 3.5 stdlib: lib/python3.5/queue.py
+
+        Blocks until all items in the queue have been gotten and processed.
+
+        The count of unfinished tasks goes up whenever an item is added to the
+        queue. The count goes down whenever a consumer thread calls task_done()
+        to indicate that the item was retrieved and all work on it is complete.
+        When the count of unfinished tasks drops to zero, join() unblocks.
+        """
+        with self.all_tasks_done:
+            while self.unfinished_tasks:
+                self.all_tasks_done.wait()

From a0d17b634c9883419b9150ee39fcd0d3a936b036 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 17 Nov 2016 21:49:36 +0100
Subject: [PATCH 02/26] use tox and linter

---
 .gitignore                   |  3 ++-
 .travis.yml                  | 16 +++++++++++-----
 persistent_queue/__init__.py |  1 +
 requirements_test.txt        |  6 ------
 setup.cfg                    | 19 +++++++++++++++++++
 tox.ini                      | 21 +++++++++++++++++++++
 6 files changed, 54 insertions(+), 12 deletions(-)
 delete mode 100644 requirements_test.txt
 create mode 100644 setup.cfg
 create mode 100644 tox.ini

diff --git a/.gitignore b/.gitignore
index c97db2f..b032e02 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,6 +1,7 @@
+.tox/
+.cache/
 .python-version
 dist
 python_persistent_queue.egg-info
 *.queue
 *.coverage
-.cache
diff --git a/.travis.yml b/.travis.yml
index 0b1e90a..43b5678 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,12 +1,18 @@
 sudo: false
 language: python
 cache: pip
-python:
-  - "3.4"
-  - "3.5"
+matrix:
+  fast_finish: true
+  include:
+    - python: 3.5
+      env: TOXENV=lint
+    - python: 3.4
+      env: env: TOXENV=py34
+    - python: 3.5
+      env: env: TOXENV=py35
 install:
-  - "pip install -r requirements_test.txt"
+  - "pip install tox"
 script:
-  - "py.test -v --cov persistent_queue"
+  - "tox"
 after_success:
   - coveralls
diff --git a/persistent_queue/__init__.py b/persistent_queue/__init__.py
index 59f02dd..b94d4a7 100644
--- a/persistent_queue/__init__.py
+++ b/persistent_queue/__init__.py
@@ -10,6 +10,7 @@
 import struct
 import threading
 import uuid
+import queue
 
 LENGTH_STRUCT = 'I'
 HEADER_STRUCT = 'II'
diff --git a/requirements_test.txt b/requirements_test.txt
deleted file mode 100644
index d872458..0000000
--- a/requirements_test.txt
+++ /dev/null
@@ -1,6 +0,0 @@
-bson==0.4.3
-coveralls>=1.1
-dill==0.2.5
-flake8==3.0.4
-pytest==3.0.1
-pytest-cov==2.3.1
diff --git a/setup.cfg b/setup.cfg
new file mode 100644
index 0000000..48513d1
--- /dev/null
+++ b/setup.cfg
@@ -0,0 +1,19 @@
+[flake8]
+max-line-length = 140
+max-complexity = 25
+ignore = E251,C901,W503
+exclude = tests/*
+addons = file,open,basestring,xrange,unicode,long,cmp
+
+[tool:pytest]
+testpaths = tests
+addopts = --capture=no --color=yes
+
+[coverage:run]
+branch = False
+
+[coverage:report]
+show_missing = True
+exclude_lines =
+    pragma: no cover
+    raise NotImplementedError()
diff --git a/tox.ini b/tox.ini
new file mode 100644
index 0000000..708565f
--- /dev/null
+++ b/tox.ini
@@ -0,0 +1,21 @@
+[tox]
+envlist = py34, py35, lint
+skipsdist = True
+toxworkdir = {env:TOX_WORK_DIR:.tox}
+
+[testenv]
+deps =
+  coveralls>=1.1
+  bson==0.4.3
+  dill==0.2.5
+  pytest==3.0.1
+  pytest-cov==2.3.1
+passenv = CODECOV_TOKEN CI CI_* TRAVIS TRAVIS_*
+setenv = HOME = {envtmpdir}
+commands = py.test -v --cov persistent_queue {posargs}
+
+[testenv:lint]
+deps =
+  flake8==3.0.4
+commands =
+  flake8 --count persistent_queue

From 65c4ed4df8ff4fe1f5adea43fa590b230abc2519 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 17 Nov 2016 22:01:34 +0100
Subject: [PATCH 03/26] restructure into file

---
 persistent_queue/__init__.py         | 422 +--------------------------
 persistent_queue/persistent_queue.py | 420 ++++++++++++++++++++++++++
 tests/test_persistent_queue.py       |   2 +-
 3 files changed, 425 insertions(+), 419 deletions(-)
 create mode 100644 persistent_queue/persistent_queue.py

diff --git a/persistent_queue/__init__.py b/persistent_queue/__init__.py
index b94d4a7..a79ae19 100644
--- a/persistent_queue/__init__.py
+++ b/persistent_queue/__init__.py
@@ -1,419 +1,5 @@
-"""
-An implementation of a persistent queue. It is optimized for peeking at values
-and then deleting them off to top of the queue.
-"""
+from .persistent_queue import PersistentQueue
 
-import logging
-import os.path
-import pickle
-import shutil
-import struct
-import threading
-import uuid
-import queue
-
-LENGTH_STRUCT = 'I'
-HEADER_STRUCT = 'II'
-START_OFFSET = 4 + 4
-
-_LOGGER = logging.getLogger(__name__)
-
-
-class PersistentQueue:
-    def __init__(self, filename, path='.', dumps=pickle.dumps,
-                 loads=pickle.loads, flush_limit=1048576):
-        self.filename = filename
-        self.path = path
-        self.file = self._open_file()
-        self.loads = loads
-        self.dumps = dumps
-        self.flush_limit = flush_limit
-        self.file_lock = threading.RLock()
-        self.pop_lock = threading.RLock()
-        self.pushed_event = threading.Event()
-
-        self.all_tasks_done = threading.Condition()
-        self.unfinished_tasks = 0
-
-        self.file.seek(0, 0)
-        self.length = struct.unpack(HEADER_STRUCT[0], self.file.read(4))[0]
-
-    def _open_file(self, mode=None):
-        filename = os.path.join(self.path, self.filename)
-
-        mode = mode or 'r+b' if os.path.isfile(filename) else 'w+b'
-        file = open(filename, mode=mode, buffering=0)
-
-        if mode == 'w+b':
-            # write length and start pointer
-            file.write(struct.pack(HEADER_STRUCT, 0, START_OFFSET))
-
-        return file
-
-    def _update_length(self, length):
-        current_pos = self.file.tell()
-
-        self.file.seek(0, 0)  # Go to the beginning of the file
-        self.file.write(struct.pack(HEADER_STRUCT[0], length))
-        self.file.flush()  # Probably not necessary since buffering=0
-        os.fsync(self.file.fileno())
-
-        self.length = length
-
-        self.file.seek(current_pos, 0)
-
-    def _get_queue_top(self):
-        current_pos = self.file.tell()
-
-        self.file.seek(START_OFFSET - 4, 0)  # Start at beginning of file
-        pos = struct.unpack(HEADER_STRUCT[1], self.file.read(4))[0]
-
-        self.file.seek(current_pos, 0)
-        return pos
-
-    def _set_queue_top(self, top):
-        current_pos = self.file.tell()
-
-        self.file.seek(START_OFFSET - 4, 0)  # Start at beginning of file
-        self.file.write(struct.pack(HEADER_STRUCT[1], top))
-        self.file.flush()  # Probably not necessary since buffering=0
-        os.fsync(self.file.fileno())
-
-        self.file.seek(current_pos, 0)
-
-    def clear(self):
-        """Removes all elements from queue."""
-        _LOGGER.debug("Clearing the queue")
-        with self.file_lock, self.pop_lock:
-            self.file.close()
-            self.file = self._open_file(mode='w+b')
-            self.length = 0
-            _LOGGER.debug("The queue has been cleared")
-
-    def copy(self, new_filename, path=None):
-        """Copies a queue to a new queue."""
-        old = os.path.join(self.path, self.filename)
-        new = os.path.join(path or self.path, new_filename)
-        shutil.copy2(old, new)
-
-        return PersistentQueue(filename=new_filename,
-                               path=path or self.path,
-                               flush_limit=self.flush_limit)
-
-    def count(self):
-        """Return the size of the queue."""
-        return self.length
-
-    def flush(self):
-        """
-        Removes elements that have been deleted or popped from the queue. This
-        will be taken care of by pop and delete.
-        """
-        _LOGGER.debug("Flushing the queue")
-
-        with self.file_lock:
-            pos = self._get_queue_top()
-
-        if pos < self.flush_limit:
-            # Ignore if the file isn't big enough -- it's not worth it
-            _LOGGER.debug("Ignoring flush because we haven't met the limit")
-            return
-
-        # Make a new file
-        random = str(uuid.uuid4()).replace('-', '')
-        temp_filename = os.path.join(self.path, self.filename + '-' + random)
-        new_file = open(temp_filename, mode='w+b', buffering=0)
-
-        # From this point on, the file can not change
-        with self.file_lock, self.pop_lock:
-            # Make sure everything is to disk
-            self.file.flush()
-            os.fsync(self.file.fileno())
-
-            start = self._get_queue_top()  # Get it again in case it changed
-            self.file.seek(0, 2)  # Go to end of file
-            end = self.file.tell()
-
-            _LOGGER.debug("Writing data to new file")
-            # Copy over meta data
-            new_file.write(struct.pack(HEADER_STRUCT,
-                                       self.count(),
-                                       START_OFFSET))
-
-            # Copy over data
-            # Do it in chunks so we aren't loading tons of data into memory
-            self.file.seek(start, 0)
-            bytes_read = 0
-            chunk_size = 4096
-            while bytes_read < end - start:
-                bytes_read += chunk_size
-                new_file.write(self.file.read(bytes_read))
-
-            new_file.flush()  # Probably not necessary since buffering=0
-            os.fsync(new_file.fileno())
-            new_file.close()
-            self.file.close()
-
-            # So far everything above this point has been safe. If something
-            # crashed, the data would still be preserved. Now we are entering
-            # the danger zone.
-
-            _LOGGER.debug("Replacing old file with new file")
-            os.replace(temp_filename, os.path.join(self.path, self.filename))
-            self.file = self._open_file()
-
-            _LOGGER.debug("Finished flushing the queue")
-
-    def pop(self, items=1, blocking=False):
-        """
-        Removes and returns a certain amount of items from the queue. If items
-        is greater than one, a list is returned.
-        """
-        _LOGGER.debug("Popping %s items", items)
-
-        # Ignore requests for zero items
-        if items == 0:
-            _LOGGER.debug("Returning empty list")
-            return []
-
-        with self.pop_lock:
-            data, queue_top = self._peek(items, blocking)
-
-            with self.file_lock:
-                self._set_queue_top(queue_top)
-
-                if isinstance(data, list):
-                    if len(data) > 0:
-                        self._update_length(self.count() - len(data))
-                elif data is not None:
-                    self._update_length(self.count() - 1)
-
-                _LOGGER.debug("Returning data from the pop")
-                return data
-
-    def _peek(self, items=1, blocking=False):
-        """
-        Returns a certain amount of items from the queue. If items is greater
-        than one, a list is returned.
-        """
-        def read_data():
-            length = struct.unpack(LENGTH_STRUCT, self.file.read(4))[0]
-            data = self.file.read(length)
-            return self.loads(data)
-
-        _LOGGER.debug("Peeking %s items", items)
-
-        # Ignore requests for zero items
-        if items == 0:
-            _LOGGER.debug("Returning empty list")
-            return [], self.file.tell()
-
-        if blocking:
-            while self.count() < items:
-                self.pushed_event.wait()
-                self.pushed_event.clear()
-
-        with self.file_lock:
-            self.file.seek(self._get_queue_top(), 0)  # Beginning of data
-            total_items = self.count() if items > self.count() else items
-            data = [read_data() for i in range(total_items)]
-            queue_top = self.file.tell()
-
-        if items == 1:
-            if len(data) == 0:
-                _LOGGER.debug("No items to peek at so returning None")
-                return None, queue_top
-            else:
-                _LOGGER.debug("Returning data from peek")
-                return data[0], queue_top
-        else:
-            _LOGGER.debug("Returning data from peek")
-            return data, queue_top
-
-    def peek(self, items=1, blocking=False):
-        with self.pop_lock:
-            return self._peek(items, blocking)[0]
-
-    def delete(self, items=1):
-        """Removes items from queue. Nothing is returned."""
-        def read_length():
-            length = struct.unpack(LENGTH_STRUCT, self.file.read(4))[0]
-            self.file.seek(length, 1)
-
-        _LOGGER.debug("Deleting %s items", items)
-
-        # Ignore requests for zero items
-        if items == 0:
-            _LOGGER.debug("Ignoring request to delete")
-            return
-
-        with self.file_lock, self.pop_lock:
-            self.file.seek(self._get_queue_top(), 0)  # Beginning of data
-            total_items = self.count() if items > self.count() else items
-
-            for _ in range(total_items):
-                read_length()
-
-            self._set_queue_top(self.file.tell())
-            self._update_length(self.count() - total_items)
-
-        _LOGGER.debug("Done deleting data")
-
-    def push(self, items):
-        """Add items to the queue."""
-        def write_data(item):
-            data = self.dumps(item)
-            self.file.write(struct.pack(LENGTH_STRUCT, len(data)))
-            self.file.write(data)
-            self.file.flush()  # Probably not necessary since buffering=0
-            os.fsync(self.file.fileno())
-
-        if not isinstance(items, list):
-            items = [items]
-
-        _LOGGER.debug("Pushing %s items", len(items))
-
-        # Ignore requests for adding zero items
-        if len(items) == 0:
-            _LOGGER.debug("Pushing zero items, ignoring request")
-            return
-
-        with self.file_lock:
-            self.file.seek(0, 2)  # Go to end of file
-
-            for i in items:
-                write_data(i)
-
-            self._update_length(self.count() + len(items))
-            self.unfinished_tasks += 1
-
-        self.pushed_event.set()
-        _LOGGER.debug("Done pushing data")
-
-    def __len__(self):
-        """Get size of queue."""
-        return self.count()
-
-    def qsize(self):
-        """
-        Provides compatibility with stdlib Queue objects.
-
-        Return the approximate size of the queue. Note, qsize() > 0 doesn’t
-        guarantee that a subsequent get() will not block, nor will qsize() <
-        maxsize guarantee that put() will not block.
-        """
-        return self.count()
-
-    def empty(self):
-        """
-        Provides compatibility with stdlib Queue objects.
-
-        Return True if the queue is empty, False otherwise. If empty() returns
-        True it doesn’t guarantee that a subsequent call to put() will not
-        block. Similarly, if empty() returns False it doesn’t guarantee that a
-        subsequent call to get() will not block.
-        """
-        return self.count() == 0
-
-    def full(self):
-        """
-        Provides compatibility with stdlib Queue objects.
-
-        Return True if the queue is full, False otherwise. If full() returns
-        True it doesn’t guarantee that a subsequent call to get() will not
-        block. Similarly, if full() returns False it doesn’t guarantee that a
-        subsequent call to put() will not block.
-        """
-        return False
-
-    def put(self, item, block=True, timeout=None):
-        """
-        Provides compatibility with stdlib Queue objects.
-        The arguments block and timeout are ignored.
-
-        Put item into the queue. If optional args block is true and timeout is
-        None (the default), block if necessary until a free slot is available.
-        If timeout is a positive number, it blocks at most timeout seconds and
-        raises the Full exception if no free slot was available within that
-        time. Otherwise (block is false), put an item on the queue if a free
-        slot is immediately available, else raise the Full exception (timeout
-        is ignored in that case).
-        """
-        self.push(item)
-
-    def put_nowait(self, item):
-        """
-        Provides compatibility with stdlib Queue objects.
-
-        Equivalent to put(item, False).
-        """
-        self.put(item, block=False)
-
-    def get(self, block=True, timeout=None):
-        """
-        Provides compatibility with stdlib Queue objects.
-        The argument timeout is ignored.
-
-        Remove and return an item from the queue. If optional args block is
-        true and timeout is None (the default), block if necessary until an
-        item is available. If timeout is a positive number, it blocks at most
-        timeout seconds and raises the Empty exception if no item was available
-        within that time. Otherwise (block is false), return an item if one is
-        immediately available, else raise the Empty exception (timeout is
-        ignored in that case).
-        """
-        item = self.pop(blocking=block)
-        if item is None:
-            raise queue.Empty
-        return item
-
-    def get_nowait(self):
-        """
-        Provides compatibility with stdlib Queue objects.
-
-        Equivalent to get(False).
-
-        Two methods are offered to support tracking whether enqueued tasks
-        have been fully processed by daemon consumer threads.
-        """
-        return self.get(block=False)
-
-    def task_done(self):
-        """
-        Provides compatibility with stdlib Queue objects.
-        Taken from the Python 3.5 stdlib: lib/python3.5/queue.py
-
-        Indicate that a formerly enqueued task is complete. Used by queue
-        consumer threads. For each get() used to fetch a task, a subsequent
-        call to task_done() tells the queue that the processing on the task is
-        complete.
-
-        If a join() is currently blocking, it will resume when all items have
-        been processed (meaning that a task_done() call was received for every
-        item that had been put() into the queue).
-
-        Raises a ValueError if called more times than there were items placed in the queue.
-        """
-        with self.all_tasks_done:
-            unfinished = self.unfinished_tasks - 1
-            if unfinished <= 0:
-                if unfinished < 0:
-                    raise ValueError('task_done() called too many times')
-                self.all_tasks_done.notify_all()
-            self.unfinished_tasks = unfinished
-
-    def join(self):
-        """
-        Provides compatibility with stdlib Queue objects.
-        Taken from the Python 3.5 stdlib: lib/python3.5/queue.py
-
-        Blocks until all items in the queue have been gotten and processed.
-
-        The count of unfinished tasks goes up whenever an item is added to the
-        queue. The count goes down whenever a consumer thread calls task_done()
-        to indicate that the item was retrieved and all work on it is complete.
-        When the count of unfinished tasks drops to zero, join() unblocks.
-        """
-        with self.all_tasks_done:
-            while self.unfinished_tasks:
-                self.all_tasks_done.wait()
+__all__ = [
+    'PersistentQueue',
+]
diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py
new file mode 100644
index 0000000..c7b52e8
--- /dev/null
+++ b/persistent_queue/persistent_queue.py
@@ -0,0 +1,420 @@
+"""
+An implementation of a persistent queue. It is optimized for peeking at values
+and then deleting them off to top of the queue.
+"""
+
+import logging
+import os.path
+import pickle
+import shutil
+import struct
+import threading
+import uuid
+import queue
+
+LENGTH_STRUCT = 'I'
+HEADER_STRUCT = 'II'
+START_OFFSET = 4 + 4
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class PersistentQueue:
+    def __init__(self, filename, path='.', dumps=pickle.dumps,
+                 loads=pickle.loads, flush_limit=1048576):
+        self.filename = filename
+        self.path = path
+        self.dumps = dumps
+        self.loads = loads
+        self.flush_limit = flush_limit
+
+        self.file = self._open_file()
+        self.file_lock = threading.RLock()
+        self.pop_lock = threading.RLock()
+        self.pushed_event = threading.Event()
+
+        self.all_tasks_done = threading.Condition()
+        self.unfinished_tasks = 0
+
+        self.file.seek(0, 0)
+        self.length = struct.unpack(HEADER_STRUCT[0], self.file.read(4))[0]
+
+    def _open_file(self, mode=None):
+        filename = os.path.join(self.path, self.filename)
+
+        mode = mode or 'r+b' if os.path.isfile(filename) else 'w+b'
+        file = open(filename, mode=mode, buffering=0)
+
+        if mode == 'w+b':
+            # write length and start pointer
+            file.write(struct.pack(HEADER_STRUCT, 0, START_OFFSET))
+
+        return file
+
+    def _update_length(self, length):
+        current_pos = self.file.tell()
+
+        self.file.seek(0, 0)  # Go to the beginning of the file
+        self.file.write(struct.pack(HEADER_STRUCT[0], length))
+        self.file.flush()  # Probably not necessary since buffering=0
+        os.fsync(self.file.fileno())
+
+        self.length = length
+
+        self.file.seek(current_pos, 0)
+
+    def _get_queue_top(self):
+        current_pos = self.file.tell()
+
+        self.file.seek(START_OFFSET - 4, 0)  # Start at beginning of file
+        pos = struct.unpack(HEADER_STRUCT[1], self.file.read(4))[0]
+
+        self.file.seek(current_pos, 0)
+        return pos
+
+    def _set_queue_top(self, top):
+        current_pos = self.file.tell()
+
+        self.file.seek(START_OFFSET - 4, 0)  # Start at beginning of file
+        self.file.write(struct.pack(HEADER_STRUCT[1], top))
+        self.file.flush()  # Probably not necessary since buffering=0
+        os.fsync(self.file.fileno())
+
+        self.file.seek(current_pos, 0)
+
+    def clear(self):
+        """Removes all elements from queue."""
+        _LOGGER.debug("Clearing the queue")
+        with self.file_lock, self.pop_lock:
+            self.file.close()
+            self.file = self._open_file(mode='w+b')
+            self.length = 0
+            _LOGGER.debug("The queue has been cleared")
+
+    def copy(self, new_filename, path=None):
+        """Copies a queue to a new queue."""
+        old = os.path.join(self.path, self.filename)
+        new = os.path.join(path or self.path, new_filename)
+        shutil.copy2(old, new)
+
+        return PersistentQueue(filename=new_filename,
+                               path=path or self.path,
+                               flush_limit=self.flush_limit)
+
+    def count(self):
+        """Return the size of the queue."""
+        return self.length
+
+    def flush(self):
+        """
+        Removes elements that have been deleted or popped from the queue. This
+        will be taken care of by pop and delete.
+        """
+        _LOGGER.debug("Flushing the queue")
+
+        with self.file_lock:
+            pos = self._get_queue_top()
+
+        if pos < self.flush_limit:
+            # Ignore if the file isn't big enough -- it's not worth it
+            _LOGGER.debug("Ignoring flush because we haven't met the limit")
+            return
+
+        # Make a new file
+        random = str(uuid.uuid4()).replace('-', '')
+        temp_filename = os.path.join(self.path, self.filename + '-' + random)
+        new_file = open(temp_filename, mode='w+b', buffering=0)
+
+        # From this point on, the file can not change
+        with self.file_lock, self.pop_lock:
+            # Make sure everything is to disk
+            self.file.flush()
+            os.fsync(self.file.fileno())
+
+            start = self._get_queue_top()  # Get it again in case it changed
+            self.file.seek(0, 2)  # Go to end of file
+            end = self.file.tell()
+
+            _LOGGER.debug("Writing data to new file")
+            # Copy over meta data
+            new_file.write(struct.pack(HEADER_STRUCT,
+                                       self.count(),
+                                       START_OFFSET))
+
+            # Copy over data
+            # Do it in chunks so we aren't loading tons of data into memory
+            self.file.seek(start, 0)
+            bytes_read = 0
+            chunk_size = 4096
+            while bytes_read < end - start:
+                bytes_read += chunk_size
+                new_file.write(self.file.read(bytes_read))
+
+            new_file.flush()  # Probably not necessary since buffering=0
+            os.fsync(new_file.fileno())
+            new_file.close()
+            self.file.close()
+
+            # So far everything above this point has been safe. If something
+            # crashed, the data would still be preserved. Now we are entering
+            # the danger zone.
+
+            _LOGGER.debug("Replacing old file with new file")
+            os.replace(temp_filename, os.path.join(self.path, self.filename))
+            self.file = self._open_file()
+
+            _LOGGER.debug("Finished flushing the queue")
+
+    def pop(self, items=1, blocking=False):
+        """
+        Removes and returns a certain amount of items from the queue. If items
+        is greater than one, a list is returned.
+        """
+        _LOGGER.debug("Popping %s items", items)
+
+        # Ignore requests for zero items
+        if items == 0:
+            _LOGGER.debug("Returning empty list")
+            return []
+
+        with self.pop_lock:
+            data, queue_top = self._peek(items, blocking)
+
+            with self.file_lock:
+                self._set_queue_top(queue_top)
+
+                if isinstance(data, list):
+                    if len(data) > 0:
+                        self._update_length(self.count() - len(data))
+                elif data is not None:
+                    self._update_length(self.count() - 1)
+
+                _LOGGER.debug("Returning data from the pop")
+                return data
+
+    def _peek(self, items=1, blocking=False):
+        """
+        Returns a certain amount of items from the queue. If items is greater
+        than one, a list is returned.
+        """
+        def read_data():
+            length = struct.unpack(LENGTH_STRUCT, self.file.read(4))[0]
+            data = self.file.read(length)
+            return self.loads(data)
+
+        _LOGGER.debug("Peeking %s items", items)
+
+        # Ignore requests for zero items
+        if items == 0:
+            _LOGGER.debug("Returning empty list")
+            return [], self.file.tell()
+
+        if blocking:
+            while self.count() < items:
+                self.pushed_event.wait()
+                self.pushed_event.clear()
+
+        with self.file_lock:
+            self.file.seek(self._get_queue_top(), 0)  # Beginning of data
+            total_items = self.count() if items > self.count() else items
+            data = [read_data() for i in range(total_items)]
+            queue_top = self.file.tell()
+
+        if items == 1:
+            if len(data) == 0:
+                _LOGGER.debug("No items to peek at so returning None")
+                return None, queue_top
+            else:
+                _LOGGER.debug("Returning data from peek")
+                return data[0], queue_top
+        else:
+            _LOGGER.debug("Returning data from peek")
+            return data, queue_top
+
+    def peek(self, items=1, blocking=False):
+        with self.pop_lock:
+            return self._peek(items, blocking)[0]
+
+    def delete(self, items=1):
+        """Removes items from queue. Nothing is returned."""
+        def read_length():
+            length = struct.unpack(LENGTH_STRUCT, self.file.read(4))[0]
+            self.file.seek(length, 1)
+
+        _LOGGER.debug("Deleting %s items", items)
+
+        # Ignore requests for zero items
+        if items == 0:
+            _LOGGER.debug("Ignoring request to delete")
+            return
+
+        with self.file_lock, self.pop_lock:
+            self.file.seek(self._get_queue_top(), 0)  # Beginning of data
+            total_items = self.count() if items > self.count() else items
+
+            for _ in range(total_items):
+                read_length()
+
+            self._set_queue_top(self.file.tell())
+            self._update_length(self.count() - total_items)
+
+        _LOGGER.debug("Done deleting data")
+
+    def push(self, items):
+        """Add items to the queue."""
+        def write_data(item):
+            data = self.dumps(item)
+            self.file.write(struct.pack(LENGTH_STRUCT, len(data)))
+            self.file.write(data)
+            self.file.flush()  # Probably not necessary since buffering=0
+            os.fsync(self.file.fileno())
+
+        if not isinstance(items, list):
+            items = [items]
+
+        _LOGGER.debug("Pushing %s items", len(items))
+
+        # Ignore requests for adding zero items
+        if len(items) == 0:
+            _LOGGER.debug("Pushing zero items, ignoring request")
+            return
+
+        with self.file_lock:
+            self.file.seek(0, 2)  # Go to end of file
+
+            for i in items:
+                write_data(i)
+
+            self._update_length(self.count() + len(items))
+            self.unfinished_tasks += 1
+
+        self.pushed_event.set()
+        _LOGGER.debug("Done pushing data")
+
+    def __len__(self):
+        """Get size of queue."""
+        return self.count()
+
+    def qsize(self):
+        """
+        Provides compatibility with stdlib Queue objects.
+
+        Return the approximate size of the queue. Note, qsize() > 0 doesn’t
+        guarantee that a subsequent get() will not block, nor will qsize() <
+        maxsize guarantee that put() will not block.
+        """
+        return self.count()
+
+    def empty(self):
+        """
+        Provides compatibility with stdlib Queue objects.
+
+        Return True if the queue is empty, False otherwise. If empty() returns
+        True it doesn’t guarantee that a subsequent call to put() will not
+        block. Similarly, if empty() returns False it doesn’t guarantee that a
+        subsequent call to get() will not block.
+        """
+        return self.count() == 0
+
+    def full(self):
+        """
+        Provides compatibility with stdlib Queue objects.
+
+        Return True if the queue is full, False otherwise. If full() returns
+        True it doesn’t guarantee that a subsequent call to get() will not
+        block. Similarly, if full() returns False it doesn’t guarantee that a
+        subsequent call to put() will not block.
+        """
+        return False
+
+    def put(self, item, block=True, timeout=None):
+        """
+        Provides compatibility with stdlib Queue objects.
+        The arguments block and timeout are ignored.
+
+        Put item into the queue. If optional args block is true and timeout is
+        None (the default), block if necessary until a free slot is available.
+        If timeout is a positive number, it blocks at most timeout seconds and
+        raises the Full exception if no free slot was available within that
+        time. Otherwise (block is false), put an item on the queue if a free
+        slot is immediately available, else raise the Full exception (timeout
+        is ignored in that case).
+        """
+        self.push(item)
+
+    def put_nowait(self, item):
+        """
+        Provides compatibility with stdlib Queue objects.
+
+        Equivalent to put(item, False).
+        """
+        self.put(item, block=False)
+
+    def get(self, block=True, timeout=None):
+        """
+        Provides compatibility with stdlib Queue objects.
+        The argument timeout is ignored.
+
+        Remove and return an item from the queue. If optional args block is
+        true and timeout is None (the default), block if necessary until an
+        item is available. If timeout is a positive number, it blocks at most
+        timeout seconds and raises the Empty exception if no item was available
+        within that time. Otherwise (block is false), return an item if one is
+        immediately available, else raise the Empty exception (timeout is
+        ignored in that case).
+        """
+        item = self.pop(blocking=block)
+        if item is None:
+            raise queue.Empty
+        return item
+
+    def get_nowait(self):
+        """
+        Provides compatibility with stdlib Queue objects.
+
+        Equivalent to get(False).
+
+        Two methods are offered to support tracking whether enqueued tasks
+        have been fully processed by daemon consumer threads.
+        """
+        return self.get(block=False)
+
+    def task_done(self):
+        """
+        Provides compatibility with stdlib Queue objects.
+        Taken from the Python 3.5 stdlib: lib/python3.5/queue.py
+
+        Indicate that a formerly enqueued task is complete. Used by queue
+        consumer threads. For each get() used to fetch a task, a subsequent
+        call to task_done() tells the queue that the processing on the task is
+        complete.
+
+        If a join() is currently blocking, it will resume when all items have
+        been processed (meaning that a task_done() call was received for every
+        item that had been put() into the queue).
+
+        Raises a ValueError if called more times than there were items placed in the queue.
+        """
+        with self.all_tasks_done:
+            unfinished = self.unfinished_tasks - 1
+            if unfinished <= 0:
+                if unfinished < 0:
+                    raise ValueError('task_done() called too many times')
+                self.all_tasks_done.notify_all()
+            self.unfinished_tasks = unfinished
+
+    def join(self):
+        """
+        Provides compatibility with stdlib Queue objects.
+        Taken from the Python 3.5 stdlib: lib/python3.5/queue.py
+
+        Blocks until all items in the queue have been gotten and processed.
+
+        The count of unfinished tasks goes up whenever an item is added to the
+        queue. The count goes down whenever a consumer thread calls task_done()
+        to indicate that the item was retrieved and all work on it is complete.
+        When the count of unfinished tasks drops to zero, join() unblocks.
+        """
+        with self.all_tasks_done:
+            while self.unfinished_tasks:
+                self.all_tasks_done.wait()
diff --git a/tests/test_persistent_queue.py b/tests/test_persistent_queue.py
index 1090702..16fed06 100644
--- a/tests/test_persistent_queue.py
+++ b/tests/test_persistent_queue.py
@@ -11,7 +11,7 @@
 class TestPersistentQueue(unittest.TestCase):
     def setUp(self):
         random = str(uuid.uuid4()).replace('-', '')
-        self.filename = '{}_{}'.format(self.id(), random)
+        self.filename = '{}_{}.queue'.format(self.id(), random)
         self.queue = PersistentQueue(self.filename)
 
     def tearDown(self):

From df13de7eb78550fb87e3d1bb881189e6be442db7 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 17 Nov 2016 22:16:23 +0100
Subject: [PATCH 04/26] mark internal members as private

---
 persistent_queue/persistent_queue.py | 170 +++++++++++++--------------
 tests/test_persistent_queue.py       |  24 ++--
 2 files changed, 96 insertions(+), 98 deletions(-)

diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py
index c7b52e8..8cf819b 100644
--- a/persistent_queue/persistent_queue.py
+++ b/persistent_queue/persistent_queue.py
@@ -28,16 +28,16 @@ def __init__(self, filename, path='.', dumps=pickle.dumps,
         self.loads = loads
         self.flush_limit = flush_limit
 
-        self.file = self._open_file()
-        self.file_lock = threading.RLock()
-        self.pop_lock = threading.RLock()
-        self.pushed_event = threading.Event()
+        self._file = self._open_file()
+        self._file_lock = threading.RLock()
+        self._pop_lock = threading.RLock()
+        self._pushed_event = threading.Event()
 
-        self.all_tasks_done = threading.Condition()
-        self.unfinished_tasks = 0
+        self._all_tasks_done = threading.Condition()
+        self._unfinished_tasks = 0
 
-        self.file.seek(0, 0)
-        self.length = struct.unpack(HEADER_STRUCT[0], self.file.read(4))[0]
+        self._file.seek(0, 0)
+        self._length = struct.unpack(HEADER_STRUCT[0], self._file.read(4))[0]
 
     def _open_file(self, mode=None):
         filename = os.path.join(self.path, self.filename)
@@ -52,47 +52,49 @@ def _open_file(self, mode=None):
         return file
 
     def _update_length(self, length):
-        current_pos = self.file.tell()
+        current_pos = self._file.tell()
 
-        self.file.seek(0, 0)  # Go to the beginning of the file
-        self.file.write(struct.pack(HEADER_STRUCT[0], length))
-        self.file.flush()  # Probably not necessary since buffering=0
-        os.fsync(self.file.fileno())
+        self._file.seek(0, 0)  # Go to the beginning of the file
+        self._file.write(struct.pack(HEADER_STRUCT[0], length))
+        self._file.flush()  # Probably not necessary since buffering=0
+        os.fsync(self._file.fileno())
 
-        self.length = length
+        self._length = length
 
-        self.file.seek(current_pos, 0)
+        self._file.seek(current_pos, 0)
 
     def _get_queue_top(self):
-        current_pos = self.file.tell()
+        current_pos = self._file.tell()
 
-        self.file.seek(START_OFFSET - 4, 0)  # Start at beginning of file
-        pos = struct.unpack(HEADER_STRUCT[1], self.file.read(4))[0]
+        self._file.seek(START_OFFSET - 4, 0)  # Start at beginning of file
+        pos = struct.unpack(HEADER_STRUCT[1], self._file.read(4))[0]
 
-        self.file.seek(current_pos, 0)
+        self._file.seek(current_pos, 0)
         return pos
 
     def _set_queue_top(self, top):
-        current_pos = self.file.tell()
+        current_pos = self._file.tell()
 
-        self.file.seek(START_OFFSET - 4, 0)  # Start at beginning of file
-        self.file.write(struct.pack(HEADER_STRUCT[1], top))
-        self.file.flush()  # Probably not necessary since buffering=0
-        os.fsync(self.file.fileno())
+        self._file.seek(START_OFFSET - 4, 0)  # Start at beginning of file
+        self._file.write(struct.pack(HEADER_STRUCT[1], top))
+        self._file.flush()  # Probably not necessary since buffering=0
+        os.fsync(self._file.fileno())
 
-        self.file.seek(current_pos, 0)
+        self._file.seek(current_pos, 0)
 
     def clear(self):
         """Removes all elements from queue."""
         _LOGGER.debug("Clearing the queue")
-        with self.file_lock, self.pop_lock:
-            self.file.close()
-            self.file = self._open_file(mode='w+b')
-            self.length = 0
+        with self._file_lock, self._pop_lock:
+            self._file.close()
+            self._file = self._open_file(mode='w+b')
+            self._length = 0
             _LOGGER.debug("The queue has been cleared")
 
     def copy(self, new_filename, path=None):
-        """Copies a queue to a new queue."""
+        """
+        Copies a queue to a new queue by duplicating the underlying file.
+        """
         old = os.path.join(self.path, self.filename)
         new = os.path.join(path or self.path, new_filename)
         shutil.copy2(old, new)
@@ -101,10 +103,6 @@ def copy(self, new_filename, path=None):
                                path=path or self.path,
                                flush_limit=self.flush_limit)
 
-    def count(self):
-        """Return the size of the queue."""
-        return self.length
-
     def flush(self):
         """
         Removes elements that have been deleted or popped from the queue. This
@@ -112,7 +110,7 @@ def flush(self):
         """
         _LOGGER.debug("Flushing the queue")
 
-        with self.file_lock:
+        with self._file_lock:
             pos = self._get_queue_top()
 
         if pos < self.flush_limit:
@@ -126,34 +124,34 @@ def flush(self):
         new_file = open(temp_filename, mode='w+b', buffering=0)
 
         # From this point on, the file can not change
-        with self.file_lock, self.pop_lock:
+        with self._file_lock, self._pop_lock:
             # Make sure everything is to disk
-            self.file.flush()
-            os.fsync(self.file.fileno())
+            self._file.flush()
+            os.fsync(self._file.fileno())
 
             start = self._get_queue_top()  # Get it again in case it changed
-            self.file.seek(0, 2)  # Go to end of file
-            end = self.file.tell()
+            self._file.seek(0, 2)  # Go to end of file
+            end = self._file.tell()
 
             _LOGGER.debug("Writing data to new file")
             # Copy over meta data
             new_file.write(struct.pack(HEADER_STRUCT,
-                                       self.count(),
+                                       self._length,
                                        START_OFFSET))
 
             # Copy over data
             # Do it in chunks so we aren't loading tons of data into memory
-            self.file.seek(start, 0)
+            self._file.seek(start, 0)
             bytes_read = 0
             chunk_size = 4096
             while bytes_read < end - start:
                 bytes_read += chunk_size
-                new_file.write(self.file.read(bytes_read))
+                new_file.write(self._file.read(bytes_read))
 
             new_file.flush()  # Probably not necessary since buffering=0
             os.fsync(new_file.fileno())
             new_file.close()
-            self.file.close()
+            self._file.close()
 
             # So far everything above this point has been safe. If something
             # crashed, the data would still be preserved. Now we are entering
@@ -161,7 +159,7 @@ def flush(self):
 
             _LOGGER.debug("Replacing old file with new file")
             os.replace(temp_filename, os.path.join(self.path, self.filename))
-            self.file = self._open_file()
+            self._file = self._open_file()
 
             _LOGGER.debug("Finished flushing the queue")
 
@@ -177,17 +175,17 @@ def pop(self, items=1, blocking=False):
             _LOGGER.debug("Returning empty list")
             return []
 
-        with self.pop_lock:
+        with self._pop_lock:
             data, queue_top = self._peek(items, blocking)
 
-            with self.file_lock:
+            with self._file_lock:
                 self._set_queue_top(queue_top)
 
                 if isinstance(data, list):
                     if len(data) > 0:
-                        self._update_length(self.count() - len(data))
+                        self._update_length(self._length - len(data))
                 elif data is not None:
-                    self._update_length(self.count() - 1)
+                    self._update_length(self._length - 1)
 
                 _LOGGER.debug("Returning data from the pop")
                 return data
@@ -198,8 +196,8 @@ def _peek(self, items=1, blocking=False):
         than one, a list is returned.
         """
         def read_data():
-            length = struct.unpack(LENGTH_STRUCT, self.file.read(4))[0]
-            data = self.file.read(length)
+            length = struct.unpack(LENGTH_STRUCT, self._file.read(4))[0]
+            data = self._file.read(length)
             return self.loads(data)
 
         _LOGGER.debug("Peeking %s items", items)
@@ -207,18 +205,18 @@ def read_data():
         # Ignore requests for zero items
         if items == 0:
             _LOGGER.debug("Returning empty list")
-            return [], self.file.tell()
+            return [], self._file.tell()
 
         if blocking:
-            while self.count() < items:
-                self.pushed_event.wait()
-                self.pushed_event.clear()
+            while self._length < items:
+                self._pushed_event.wait()
+                self._pushed_event.clear()
 
-        with self.file_lock:
-            self.file.seek(self._get_queue_top(), 0)  # Beginning of data
-            total_items = self.count() if items > self.count() else items
+        with self._file_lock:
+            self._file.seek(self._get_queue_top(), 0)  # Beginning of data
+            total_items = self._length if items > self._length else items
             data = [read_data() for i in range(total_items)]
-            queue_top = self.file.tell()
+            queue_top = self._file.tell()
 
         if items == 1:
             if len(data) == 0:
@@ -232,14 +230,14 @@ def read_data():
             return data, queue_top
 
     def peek(self, items=1, blocking=False):
-        with self.pop_lock:
+        with self._pop_lock:
             return self._peek(items, blocking)[0]
 
     def delete(self, items=1):
         """Removes items from queue. Nothing is returned."""
         def read_length():
-            length = struct.unpack(LENGTH_STRUCT, self.file.read(4))[0]
-            self.file.seek(length, 1)
+            length = struct.unpack(LENGTH_STRUCT, self._file.read(4))[0]
+            self._file.seek(length, 1)
 
         _LOGGER.debug("Deleting %s items", items)
 
@@ -248,15 +246,15 @@ def read_length():
             _LOGGER.debug("Ignoring request to delete")
             return
 
-        with self.file_lock, self.pop_lock:
-            self.file.seek(self._get_queue_top(), 0)  # Beginning of data
-            total_items = self.count() if items > self.count() else items
+        with self._file_lock, self._pop_lock:
+            self._file.seek(self._get_queue_top(), 0)  # Beginning of data
+            total_items = self._length if items > self._length else items
 
             for _ in range(total_items):
                 read_length()
 
-            self._set_queue_top(self.file.tell())
-            self._update_length(self.count() - total_items)
+            self._set_queue_top(self._file.tell())
+            self._update_length(self._length - total_items)
 
         _LOGGER.debug("Done deleting data")
 
@@ -264,10 +262,10 @@ def push(self, items):
         """Add items to the queue."""
         def write_data(item):
             data = self.dumps(item)
-            self.file.write(struct.pack(LENGTH_STRUCT, len(data)))
-            self.file.write(data)
-            self.file.flush()  # Probably not necessary since buffering=0
-            os.fsync(self.file.fileno())
+            self._file.write(struct.pack(LENGTH_STRUCT, len(data)))
+            self._file.write(data)
+            self._file.flush()  # Probably not necessary since buffering=0
+            os.fsync(self._file.fileno())
 
         if not isinstance(items, list):
             items = [items]
@@ -279,21 +277,21 @@ def write_data(item):
             _LOGGER.debug("Pushing zero items, ignoring request")
             return
 
-        with self.file_lock:
-            self.file.seek(0, 2)  # Go to end of file
+        with self._file_lock:
+            self._file.seek(0, 2)  # Go to end of file
 
             for i in items:
                 write_data(i)
 
-            self._update_length(self.count() + len(items))
-            self.unfinished_tasks += 1
+            self._update_length(self._length + len(items))
+            self._unfinished_tasks += 1
 
-        self.pushed_event.set()
+        self._pushed_event.set()
         _LOGGER.debug("Done pushing data")
 
     def __len__(self):
         """Get size of queue."""
-        return self.count()
+        return self._length
 
     def qsize(self):
         """
@@ -303,7 +301,7 @@ def qsize(self):
         guarantee that a subsequent get() will not block, nor will qsize() <
         maxsize guarantee that put() will not block.
         """
-        return self.count()
+        return self._length
 
     def empty(self):
         """
@@ -314,7 +312,7 @@ def empty(self):
         block. Similarly, if empty() returns False it doesn’t guarantee that a
         subsequent call to get() will not block.
         """
-        return self.count() == 0
+        return self._length == 0
 
     def full(self):
         """
@@ -395,13 +393,13 @@ def task_done(self):
 
         Raises a ValueError if called more times than there were items placed in the queue.
         """
-        with self.all_tasks_done:
-            unfinished = self.unfinished_tasks - 1
+        with self._all_tasks_done:
+            unfinished = self._unfinished_tasks - 1
             if unfinished <= 0:
                 if unfinished < 0:
                     raise ValueError('task_done() called too many times')
-                self.all_tasks_done.notify_all()
-            self.unfinished_tasks = unfinished
+                self._all_tasks_done.notify_all()
+            self._unfinished_tasks = unfinished
 
     def join(self):
         """
@@ -415,6 +413,6 @@ def join(self):
         to indicate that the item was retrieved and all work on it is complete.
         When the count of unfinished tasks drops to zero, join() unblocks.
         """
-        with self.all_tasks_done:
-            while self.unfinished_tasks:
-                self.all_tasks_done.wait()
+        with self._all_tasks_done:
+            while self._unfinished_tasks:
+                self._all_tasks_done.wait()
diff --git a/tests/test_persistent_queue.py b/tests/test_persistent_queue.py
index 16fed06..0af9d5f 100644
--- a/tests/test_persistent_queue.py
+++ b/tests/test_persistent_queue.py
@@ -20,21 +20,21 @@ def tearDown(self):
     def test_count(self):
 
         self.assertEqual(len(self.queue), 0)
-        self.assertEqual(self.queue.count(), 0)
+        self.assertEqual(self.queue.qsize(), 0)
 
         self.queue.push(1)
         self.assertEqual(len(self.queue), 1)
-        self.assertEqual(self.queue.count(), 1)
+        self.assertEqual(self.queue.qsize(), 1)
 
         self.queue.push(2)
         self.assertEqual(len(self.queue), 2)
-        self.assertEqual(self.queue.count(), 2)
+        self.assertEqual(self.queue.qsize(), 2)
 
         for i in range(100 + 1):
             self.queue.push(i)
 
         self.assertEqual(len(self.queue), 103)
-        self.assertEqual(self.queue.count(), 103)
+        self.assertEqual(self.queue.qsize(), 103)
 
     def test_clear(self):
         self.queue.push(5)
@@ -175,26 +175,26 @@ def test_peek_no_values(self):
         self.assertEqual(self.queue.peek(), None)
 
     def test_big_file_1(self):
-        data = {"a": list(range(1000))}
+        data = {"a": list(range(500))}
 
-        for i in range(2000):
+        for i in range(1000):
             self.queue.push(data)
 
-        self.assertEqual(len(self.queue), 2000)
+        self.assertEqual(len(self.queue), 1000)
 
-        for i in range(1995):
+        for i in range(995):
             self.assertEqual(self.queue.pop(), data)
             self.queue.flush()
 
         self.assertEqual(len(self.queue), 5)
 
     def test_big_file_2(self):
-        data = {"a": list(range(1000))}
+        data = {"a": list(range(500))}
 
-        for i in range(2000):
+        for i in range(1000):
             self.queue.push(data)
 
-        self.assertEqual(self.queue.pop(1995), [data for i in range(1995)])
+        self.assertEqual(self.queue.pop(995), [data for i in range(995)])
         self.queue.flush()
         self.assertEqual(len(self.queue), 5)
 
@@ -270,7 +270,7 @@ def random_stuff():
                     for i in range(random_number % 10):
                         self.queue.push({"test": [1, 2, 3], "foo": "bar", "1": random_number})
 
-        threads = [threading.Thread(target=random_stuff) for _ in range(50)]
+        threads = [threading.Thread(target=random_stuff) for _ in range(10)]
 
         for t in threads:
             t.start()

From c395324b5bd1bc351a95aaab9c8b3671c3ad8e97 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 17 Nov 2016 22:22:41 +0100
Subject: [PATCH 05/26] reorder methods

---
 persistent_queue/persistent_queue.py | 56 +++++++++++++++++-----------
 1 file changed, 34 insertions(+), 22 deletions(-)

diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py
index 8cf819b..10e4182 100644
--- a/persistent_queue/persistent_queue.py
+++ b/persistent_queue/persistent_queue.py
@@ -82,26 +82,7 @@ def _set_queue_top(self, top):
 
         self._file.seek(current_pos, 0)
 
-    def clear(self):
-        """Removes all elements from queue."""
-        _LOGGER.debug("Clearing the queue")
-        with self._file_lock, self._pop_lock:
-            self._file.close()
-            self._file = self._open_file(mode='w+b')
-            self._length = 0
-            _LOGGER.debug("The queue has been cleared")
-
-    def copy(self, new_filename, path=None):
-        """
-        Copies a queue to a new queue by duplicating the underlying file.
-        """
-        old = os.path.join(self.path, self.filename)
-        new = os.path.join(path or self.path, new_filename)
-        shutil.copy2(old, new)
 
-        return PersistentQueue(filename=new_filename,
-                               path=path or self.path,
-                               flush_limit=self.flush_limit)
 
     def flush(self):
         """
@@ -289,9 +270,8 @@ def write_data(item):
         self._pushed_event.set()
         _LOGGER.debug("Done pushing data")
 
-    def __len__(self):
-        """Get size of queue."""
-        return self._length
+
+
 
     def qsize(self):
         """
@@ -416,3 +396,35 @@ def join(self):
         with self._all_tasks_done:
             while self._unfinished_tasks:
                 self._all_tasks_done.wait()
+
+
+
+
+    def clear(self):
+        """
+        Removes all elements from queue, by truncating the file and reloading.
+        """
+        _LOGGER.debug("Clearing the queue")
+        with self._file_lock, self._pop_lock:
+            self._file.close()
+            self._file = self._open_file(mode='w+b')
+            self._length = 0
+            _LOGGER.debug("The queue has been cleared")
+
+    def copy(self, new_filename, path=None):
+        """
+        Copies a queue to a new queue by duplicating the underlying file.
+        """
+        old = os.path.join(self.path, self.filename)
+        new = os.path.join(path or self.path, new_filename)
+        shutil.copy2(old, new)
+
+        return PersistentQueue(filename=new_filename,
+                               path=path or self.path,
+                               flush_limit=self.flush_limit)
+
+    def __len__(self):
+        """
+        Get size of queue.
+        """
+        return self._length

From 9c4152ec0399c6a0fd77dbccbeb8d9ea513075a6 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 17 Nov 2016 22:35:23 +0100
Subject: [PATCH 06/26] simplify filename and path usage

---
 persistent_queue/persistent_queue.py | 37 ++++++++++++++--------------
 1 file changed, 19 insertions(+), 18 deletions(-)

diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py
index 10e4182..cfaa990 100644
--- a/persistent_queue/persistent_queue.py
+++ b/persistent_queue/persistent_queue.py
@@ -20,10 +20,16 @@
 
 
 class PersistentQueue:
-    def __init__(self, filename, path='.', dumps=pickle.dumps,
-                 loads=pickle.loads, flush_limit=1048576):
-        self.filename = filename
-        self.path = path
+    def __init__(self, filename, dumps=pickle.dumps, loads=pickle.loads, flush_limit=1048576):
+        """
+        Creates a new PersistentQueue object and underlying file.
+
+        filename: must be a full path to the new file.
+        dumps: the function called for persisting queue items to the file.
+        loads: the function called for loading queue items from the file.
+        flush_limit: below this filesize, flush() is a no-op.
+        """
+        self.filename = os.path.abspath(filename)
         self.dumps = dumps
         self.loads = loads
         self.flush_limit = flush_limit
@@ -40,10 +46,8 @@ def __init__(self, filename, path='.', dumps=pickle.dumps,
         self._length = struct.unpack(HEADER_STRUCT[0], self._file.read(4))[0]
 
     def _open_file(self, mode=None):
-        filename = os.path.join(self.path, self.filename)
-
-        mode = mode or 'r+b' if os.path.isfile(filename) else 'w+b'
-        file = open(filename, mode=mode, buffering=0)
+        mode = mode or 'r+b' if os.path.isfile(self.filename) else 'w+b'
+        file = open(self.filename, mode=mode, buffering=0)
 
         if mode == 'w+b':
             # write length and start pointer
@@ -101,7 +105,7 @@ def flush(self):
 
         # Make a new file
         random = str(uuid.uuid4()).replace('-', '')
-        temp_filename = os.path.join(self.path, self.filename + '-' + random)
+        temp_filename = self.filename + '-' + random
         new_file = open(temp_filename, mode='w+b', buffering=0)
 
         # From this point on, the file can not change
@@ -139,7 +143,7 @@ def flush(self):
             # the danger zone.
 
             _LOGGER.debug("Replacing old file with new file")
-            os.replace(temp_filename, os.path.join(self.path, self.filename))
+            os.replace(temp_filename, self.filename)
             self._file = self._open_file()
 
             _LOGGER.debug("Finished flushing the queue")
@@ -411,17 +415,14 @@ def clear(self):
             self._length = 0
             _LOGGER.debug("The queue has been cleared")
 
-    def copy(self, new_filename, path=None):
+    def copy(self, new_filename):
         """
         Copies a queue to a new queue by duplicating the underlying file.
-        """
-        old = os.path.join(self.path, self.filename)
-        new = os.path.join(path or self.path, new_filename)
-        shutil.copy2(old, new)
 
-        return PersistentQueue(filename=new_filename,
-                               path=path or self.path,
-                               flush_limit=self.flush_limit)
+        new_filename: must be a full path to the new file.
+        """
+        shutil.copy2(self.filename, new_filename)
+        return PersistentQueue(filename=new_filename, flush_limit=self.flush_limit)
 
     def __len__(self):
         """

From 0ed3aba124c61a55b9ded696a82ffee12490b70e Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 17 Nov 2016 22:58:44 +0100
Subject: [PATCH 07/26] move methods

---
 persistent_queue/persistent_queue.py | 207 ++++++++++++---------------
 1 file changed, 93 insertions(+), 114 deletions(-)

diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py
index cfaa990..4293b61 100644
--- a/persistent_queue/persistent_queue.py
+++ b/persistent_queue/persistent_queue.py
@@ -87,94 +87,6 @@ def _set_queue_top(self, top):
         self._file.seek(current_pos, 0)
 
 
-
-    def flush(self):
-        """
-        Removes elements that have been deleted or popped from the queue. This
-        will be taken care of by pop and delete.
-        """
-        _LOGGER.debug("Flushing the queue")
-
-        with self._file_lock:
-            pos = self._get_queue_top()
-
-        if pos < self.flush_limit:
-            # Ignore if the file isn't big enough -- it's not worth it
-            _LOGGER.debug("Ignoring flush because we haven't met the limit")
-            return
-
-        # Make a new file
-        random = str(uuid.uuid4()).replace('-', '')
-        temp_filename = self.filename + '-' + random
-        new_file = open(temp_filename, mode='w+b', buffering=0)
-
-        # From this point on, the file can not change
-        with self._file_lock, self._pop_lock:
-            # Make sure everything is to disk
-            self._file.flush()
-            os.fsync(self._file.fileno())
-
-            start = self._get_queue_top()  # Get it again in case it changed
-            self._file.seek(0, 2)  # Go to end of file
-            end = self._file.tell()
-
-            _LOGGER.debug("Writing data to new file")
-            # Copy over meta data
-            new_file.write(struct.pack(HEADER_STRUCT,
-                                       self._length,
-                                       START_OFFSET))
-
-            # Copy over data
-            # Do it in chunks so we aren't loading tons of data into memory
-            self._file.seek(start, 0)
-            bytes_read = 0
-            chunk_size = 4096
-            while bytes_read < end - start:
-                bytes_read += chunk_size
-                new_file.write(self._file.read(bytes_read))
-
-            new_file.flush()  # Probably not necessary since buffering=0
-            os.fsync(new_file.fileno())
-            new_file.close()
-            self._file.close()
-
-            # So far everything above this point has been safe. If something
-            # crashed, the data would still be preserved. Now we are entering
-            # the danger zone.
-
-            _LOGGER.debug("Replacing old file with new file")
-            os.replace(temp_filename, self.filename)
-            self._file = self._open_file()
-
-            _LOGGER.debug("Finished flushing the queue")
-
-    def pop(self, items=1, blocking=False):
-        """
-        Removes and returns a certain amount of items from the queue. If items
-        is greater than one, a list is returned.
-        """
-        _LOGGER.debug("Popping %s items", items)
-
-        # Ignore requests for zero items
-        if items == 0:
-            _LOGGER.debug("Returning empty list")
-            return []
-
-        with self._pop_lock:
-            data, queue_top = self._peek(items, blocking)
-
-            with self._file_lock:
-                self._set_queue_top(queue_top)
-
-                if isinstance(data, list):
-                    if len(data) > 0:
-                        self._update_length(self._length - len(data))
-                elif data is not None:
-                    self._update_length(self._length - 1)
-
-                _LOGGER.debug("Returning data from the pop")
-                return data
-
     def _peek(self, items=1, blocking=False):
         """
         Returns a certain amount of items from the queue. If items is greater
@@ -218,33 +130,12 @@ def peek(self, items=1, blocking=False):
         with self._pop_lock:
             return self._peek(items, blocking)[0]
 
-    def delete(self, items=1):
-        """Removes items from queue. Nothing is returned."""
-        def read_length():
-            length = struct.unpack(LENGTH_STRUCT, self._file.read(4))[0]
-            self._file.seek(length, 1)
-
-        _LOGGER.debug("Deleting %s items", items)
-
-        # Ignore requests for zero items
-        if items == 0:
-            _LOGGER.debug("Ignoring request to delete")
-            return
-
-        with self._file_lock, self._pop_lock:
-            self._file.seek(self._get_queue_top(), 0)  # Beginning of data
-            total_items = self._length if items > self._length else items
-
-            for _ in range(total_items):
-                read_length()
-
-            self._set_queue_top(self._file.tell())
-            self._update_length(self._length - total_items)
-
-        _LOGGER.debug("Done deleting data")
-
     def push(self, items):
-        """Add items to the queue."""
+        """
+        Add items to the queue.
+
+        items: single element, or list
+        """
         def write_data(item):
             data = self.dumps(item)
             self._file.write(struct.pack(LENGTH_STRUCT, len(data)))
@@ -424,6 +315,94 @@ def copy(self, new_filename):
         shutil.copy2(self.filename, new_filename)
         return PersistentQueue(filename=new_filename, flush_limit=self.flush_limit)
 
+    def flush(self):
+        """
+        Removes elements that have been deleted or popped from the queue.
+        """
+        _LOGGER.debug("Flushing the queue")
+
+        with self._file_lock:
+            pos = self._get_queue_top()
+
+        if pos < self.flush_limit:
+            # Ignore if the file isn't big enough -- it's not worth it
+            _LOGGER.debug("Ignoring flush because we haven't met the limit")
+            return
+
+        # Make a new file
+        random = str(uuid.uuid4()).replace('-', '')
+        temp_filename = self.filename + '-' + random
+        new_file = open(temp_filename, mode='w+b', buffering=0)
+
+        # From this point on, the file can not change
+        with self._file_lock, self._pop_lock:
+            # Make sure everything is to disk
+            self._file.flush()
+            os.fsync(self._file.fileno())
+
+            start = self._get_queue_top()  # Get it again in case it changed
+            self._file.seek(0, 2)  # Go to end of file
+            end = self._file.tell()
+
+            _LOGGER.debug("Writing data to new file")
+            # Copy over meta data
+            new_file.write(struct.pack(HEADER_STRUCT,
+                                       self._length,
+                                       START_OFFSET))
+
+            # Copy over data
+            # Do it in chunks so we aren't loading tons of data into memory
+            self._file.seek(start, 0)
+            bytes_read = 0
+            chunk_size = 4096
+            while bytes_read < end - start:
+                bytes_read += chunk_size
+                new_file.write(self._file.read(bytes_read))
+
+            new_file.flush()  # Probably not necessary since buffering=0
+            os.fsync(new_file.fileno())
+            new_file.close()
+            self._file.close()
+
+            # So far everything above this point has been safe. If something
+            # crashed, the data would still be preserved. Now we are entering
+            # the danger zone.
+
+            _LOGGER.debug("Replacing old file with new file")
+            os.replace(temp_filename, self.filename)
+            self._file = self._open_file()
+
+            _LOGGER.debug("Finished flushing the queue")
+
+    def delete(self, items=1):
+        """
+        Removes items from queue.
+
+        items: number of how many items will be deleted
+        """
+        def read_length():
+            length = struct.unpack(LENGTH_STRUCT, self._file.read(4))[0]
+            self._file.seek(length, 1)
+
+        _LOGGER.debug("Deleting %s items", items)
+
+        # Ignore requests for zero items
+        if items == 0:
+            _LOGGER.debug("Ignoring request to delete")
+            return
+
+        with self._file_lock, self._pop_lock:
+            self._file.seek(self._get_queue_top(), 0)  # Beginning of data
+            total_items = self._length if items > self._length else items
+
+            for _ in range(total_items):
+                read_length()
+
+            self._set_queue_top(self._file.tell())
+            self._update_length(self._length - total_items)
+
+        _LOGGER.debug("Done deleting data")
+
     def __len__(self):
         """
         Get size of queue.

From ab0f6c382f415574cbe0513d87f2b0899044d3b4 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 17 Nov 2016 23:07:06 +0100
Subject: [PATCH 08/26] implement queue.get

---
 persistent_queue/persistent_queue.py | 33 +++++++++++++---
 tests/test_persistent_queue.py       | 57 +++++++++++++++-------------
 2 files changed, 57 insertions(+), 33 deletions(-)

diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py
index 4293b61..fa92a01 100644
--- a/persistent_queue/persistent_queue.py
+++ b/persistent_queue/persistent_queue.py
@@ -223,10 +223,12 @@ def put_nowait(self, item):
         """
         self.put(item, block=False)
 
-    def get(self, block=True, timeout=None):
+    def get(self, block=True, timeout=None, items=1):
         """
         Provides compatibility with stdlib Queue objects.
-        The argument timeout is ignored.
+        timeout: this argument is ignored.
+        items: number of how many items are returned. If items is greater than
+        one, a list is returned.
 
         Remove and return an item from the queue. If optional args block is
         true and timeout is None (the default), block if necessary until an
@@ -236,10 +238,29 @@ def get(self, block=True, timeout=None):
         immediately available, else raise the Empty exception (timeout is
         ignored in that case).
         """
-        item = self.pop(blocking=block)
-        if item is None:
-            raise queue.Empty
-        return item
+        _LOGGER.debug("Getting %s items", items)
+
+        # Ignore requests for zero items
+        if items == 0:
+            _LOGGER.debug("Returning empty list")
+            return []
+
+        with self._pop_lock:
+            data, queue_top = self._peek(items, block)
+
+            with self._file_lock:
+                self._set_queue_top(queue_top)
+
+                if isinstance(data, list):
+                    if len(data) > 0:
+                        self._update_length(self._length - len(data))
+                elif data is not None:
+                    self._update_length(self._length - 1)
+                elif data is None:
+                    raise queue.Empty
+
+                _LOGGER.debug("Returning data from get")
+                return data
 
     def get_nowait(self):
         """
diff --git a/tests/test_persistent_queue.py b/tests/test_persistent_queue.py
index 0af9d5f..64fba08 100644
--- a/tests/test_persistent_queue.py
+++ b/tests/test_persistent_queue.py
@@ -4,6 +4,7 @@
 import time
 import unittest
 import uuid
+import queue
 
 from persistent_queue import PersistentQueue
 
@@ -59,15 +60,15 @@ def test_push(self):
         self.queue.push([])
         self.assertEqual(self.queue.peek(5), [5, 10, 15, 20, data])
 
-    def test_pop(self):
+    def test_get(self):
         self.queue.push('a')
         self.queue.push('b')
         self.assertEqual(len(self.queue), 2)
 
-        self.assertEqual(self.queue.pop(), 'a')
+        self.assertEqual(self.queue.get(), 'a')
         self.assertEqual(len(self.queue), 1)
 
-        self.assertEqual(self.queue.pop(1), 'b')
+        self.assertEqual(self.queue.get(items=1), 'b')
         self.assertEqual(len(self.queue), 0)
 
         self.queue.push('a')
@@ -76,20 +77,20 @@ def test_pop(self):
         self.queue.push('d')
         self.assertEqual(len(self.queue), 4)
 
-        self.assertEqual(self.queue.pop(3), ['a', 'b', 'c'])
+        self.assertEqual(self.queue.get(items=3), ['a', 'b', 'c'])
         self.assertEqual(len(self.queue), 1)
 
-        self.assertEqual(self.queue.pop(100), ['d'])
+        self.assertEqual(self.queue.get(block=False, items=100), ['d'])
         self.assertEqual(len(self.queue), 0)
 
-        self.assertEqual(self.queue.pop(100), [])
-        self.assertEqual(self.queue.pop(), None)
+        self.assertEqual(self.queue.get(block=False, items=100), [])
+        self.assertEqual(len(self.queue), 0)
 
         self.queue.push('d')
-        self.assertEqual(self.queue.pop(0), [])
+        self.assertEqual(self.queue.get(items=0), [])
         self.assertEqual(len(self.queue), 1)
 
-    def test_pop_blocking(self):
+    def test_get_blocking(self):
         done = False
         def func():
             nonlocal done
@@ -101,14 +102,16 @@ def func():
         t.start()
 
         self.assertFalse(done)
-        data = self.queue.pop(blocking=True)
+        data = self.queue.get()
         self.assertTrue(done)
         self.assertEqual(data, 5)
         self.assertEqual(len(self.queue), 0)
 
-    def test_pop_no_values(self):
-        self.assertEqual(self.queue.pop(5), [])
-        self.assertEqual(self.queue.pop(), None)
+    def test_get_non_blocking_no_values(self):
+        self.assertEqual(self.queue.get(block=False, items=5), [])
+
+        with self.assertRaises(queue.Empty):
+            self.queue.get(block=False)
 
     def test_peek(self):
         self.queue.push(1)
@@ -183,7 +186,7 @@ def test_big_file_1(self):
         self.assertEqual(len(self.queue), 1000)
 
         for i in range(995):
-            self.assertEqual(self.queue.pop(), data)
+            self.assertEqual(self.queue.get(), data)
             self.queue.flush()
 
         self.assertEqual(len(self.queue), 5)
@@ -194,7 +197,7 @@ def test_big_file_2(self):
         for i in range(1000):
             self.queue.push(data)
 
-        self.assertEqual(self.queue.pop(995), [data for i in range(995)])
+        self.assertEqual(self.queue.get(items=995), [data for i in range(995)])
         self.queue.flush()
         self.assertEqual(len(self.queue), 5)
 
@@ -205,15 +208,15 @@ def test_copy(self):
         new_queue_name = 'another_queue'
         self.queue.push([5, 4, 3, 2, 1])
         self.assertEqual(len(self.queue), 5)
-        self.assertEqual(self.queue.pop(), 5)
+        self.assertEqual(self.queue.get(), 5)
 
         new_queue = self.queue.copy(new_queue_name)
 
         self.assertEqual(len(self.queue), len(new_queue))
-        self.assertEqual(self.queue.pop(), new_queue.pop())
-        self.assertEqual(self.queue.pop(), new_queue.pop())
-        self.assertEqual(self.queue.pop(), new_queue.pop())
-        self.assertEqual(self.queue.pop(), new_queue.pop())
+        self.assertEqual(self.queue.get(), new_queue.get())
+        self.assertEqual(self.queue.get(), new_queue.get())
+        self.assertEqual(self.queue.get(), new_queue.get())
+        self.assertEqual(self.queue.get(), new_queue.get())
 
         os.remove(new_queue_name)
 
@@ -229,9 +232,9 @@ def test_usage(self):
 
         self.queue.push('foobar')
 
-        self.assertEqual(self.queue.pop(), 1)
+        self.assertEqual(self.queue.get(), 1)
         self.assertEqual(len(self.queue), 6)
-        self.assertEqual(self.queue.pop(6), [2, 3, 'a', 'b', 'c', 'foobar'])
+        self.assertEqual(self.queue.get(items=6), [2, 3, 'a', 'b', 'c', 'foobar'])
 
     def test_delete(self):
         self.queue.push(2)
@@ -243,7 +246,7 @@ def test_delete(self):
         self.queue.delete(2)
         self.assertEqual(len(self.queue), 2)
         self.assertEqual(self.queue.peek(2), [7, 11])
-        self.assertEqual(self.queue.pop(2), [7, 11])
+        self.assertEqual(self.queue.get(items=2), [7, 11])
 
         self.queue.push(2)
         self.queue.delete(1000)
@@ -265,7 +268,7 @@ def random_stuff():
                 if random_number % 3 == 0:
                     self.queue.peek(random_number % 5)
                 elif random_number % 2 == 0:
-                    self.queue.pop(random_number % 5)
+                    self.queue.get(items=(random_number % 5))
                 else:
                     for i in range(random_number % 10):
                         self.queue.push({"test": [1, 2, 3], "foo": "bar", "1": random_number})
@@ -280,7 +283,7 @@ def random_stuff():
 
         # Remove everything that is left so we make sure it is serializable
         for _ in range(len(self.queue)):
-            self.queue.pop()
+            self.queue.get()
 
 
 class TestPersistentQueueWithDill(TestPersistentQueue):
@@ -314,7 +317,7 @@ def test_big_file_1(self):
         self.assertEqual(len(self.queue), 2000)
 
         for i in range(1995):
-            self.assertEqual(self.queue.pop(), data)
+            self.assertEqual(self.queue.get(), data)
             self.queue.flush()
 
         self.assertEqual(len(self.queue), 5)
@@ -325,7 +328,7 @@ def test_big_file_2(self):
         for i in range(2000):
             self.queue.push(data)
 
-        self.assertEqual(self.queue.pop(1995), [data for i in range(1995)])
+        self.assertEqual(self.queue.get(items=1995), [data for i in range(1995)])
         self.queue.flush()
         self.assertEqual(len(self.queue), 5)
 

From 86e84d6cf268867c98e14a9d0adb69e543976587 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 17 Nov 2016 23:14:35 +0100
Subject: [PATCH 09/26] adapt tests

---
 persistent_queue/persistent_queue.py | 16 +++++++++----
 tests/test_persistent_queue.py       | 34 ++++++++++++++--------------
 2 files changed, 29 insertions(+), 21 deletions(-)

diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py
index fa92a01..ae94adc 100644
--- a/persistent_queue/persistent_queue.py
+++ b/persistent_queue/persistent_queue.py
@@ -86,7 +86,6 @@ def _set_queue_top(self, top):
 
         self._file.seek(current_pos, 0)
 
-
     def _peek(self, items=1, blocking=False):
         """
         Returns a certain amount of items from the queue. If items is greater
@@ -126,9 +125,9 @@ def read_data():
             _LOGGER.debug("Returning data from peek")
             return data, queue_top
 
-    def peek(self, items=1, blocking=False):
-        with self._pop_lock:
-            return self._peek(items, blocking)[0]
+
+
+
 
     def push(self, items):
         """
@@ -240,6 +239,8 @@ def get(self, block=True, timeout=None, items=1):
         """
         _LOGGER.debug("Getting %s items", items)
 
+        # TODO: implement timeout
+
         # Ignore requests for zero items
         if items == 0:
             _LOGGER.debug("Returning empty list")
@@ -316,6 +317,13 @@ def join(self):
 
 
 
+    def peek(self, block=False, items=1):
+        """
+        Peeks into the queue and returns items without removing them.
+        """
+        with self._pop_lock:
+            return self._peek(items, block)[0]
+
     def clear(self):
         """
         Removes all elements from queue, by truncating the file and reloading.
diff --git a/tests/test_persistent_queue.py b/tests/test_persistent_queue.py
index 64fba08..a142391 100644
--- a/tests/test_persistent_queue.py
+++ b/tests/test_persistent_queue.py
@@ -41,24 +41,24 @@ def test_clear(self):
         self.queue.push(5)
         self.queue.push(50)
 
-        self.assertEqual(self.queue.peek(2), [5, 50])
+        self.assertEqual(self.queue.peek(items=2), [5, 50])
         self.assertEqual(len(self.queue), 2)
         self.queue.clear()
         self.assertEqual(len(self.queue), 0)
 
     def test_push(self):
         self.queue.push(5)
-        self.assertEqual(self.queue.peek(1), 5)
+        self.assertEqual(self.queue.peek(items=1), 5)
 
         self.queue.push([10, 15, 20])
-        self.assertEqual(self.queue.peek(4), [5, 10, 15, 20])
+        self.assertEqual(self.queue.peek(items=4), [5, 10, 15, 20])
 
         data = {"a": 1, "b": 2, "c": [1, 2, 3]}
         self.queue.push(data)
-        self.assertEqual(self.queue.peek(5), [5, 10, 15, 20, data])
+        self.assertEqual(self.queue.peek(items=5), [5, 10, 15, 20, data])
 
         self.queue.push([])
-        self.assertEqual(self.queue.peek(5), [5, 10, 15, 20, data])
+        self.assertEqual(self.queue.peek(items=5), [5, 10, 15, 20, data])
 
     def test_get(self):
         self.queue.push('a')
@@ -119,21 +119,21 @@ def test_peek(self):
         self.queue.push("test")
 
         self.assertEqual(self.queue.peek(), 1)
-        self.assertEqual(self.queue.peek(1), 1)
-        self.assertEqual(self.queue.peek(2), [1, 2])
-        self.assertEqual(self.queue.peek(3), [1, 2, "test"])
+        self.assertEqual(self.queue.peek(items=1), 1)
+        self.assertEqual(self.queue.peek(items=2), [1, 2])
+        self.assertEqual(self.queue.peek(items=3), [1, 2, "test"])
 
-        self.assertEqual(self.queue.peek(100), [1, 2, "test"])
+        self.assertEqual(self.queue.peek(items=100), [1, 2, "test"])
 
         self.queue.clear()
 
         self.queue.push(1)
         self.assertEqual(len(self.queue), 1)
         self.assertEqual(self.queue.peek(), 1)
-        self.assertEqual(self.queue.peek(1), 1)
-        self.assertEqual(self.queue.peek(2), [1])
+        self.assertEqual(self.queue.peek(items=1), 1)
+        self.assertEqual(self.queue.peek(items=2), [1])
 
-        self.assertEqual(self.queue.peek(0), [])
+        self.assertEqual(self.queue.peek(items=0), [])
 
     def test_peek_blocking(self):
         done = False
@@ -147,7 +147,7 @@ def func():
         t.start()
 
         self.assertFalse(done)
-        data = self.queue.peek(blocking=True)
+        data = self.queue.peek(block=True)
         self.assertTrue(done)
         self.assertEqual(data, 5)
         self.assertEqual(len(self.queue), 1)
@@ -167,14 +167,14 @@ def func():
         t = threading.Thread(target=func)
         t.start()
 
-        data = self.queue.peek(5, blocking=True)
+        data = self.queue.peek(items=5, block=True)
         done_peeking = True
         self.assertTrue(done_pushing)
         self.assertEqual(data, [0, 1, 2, 3, 4])
         self.assertEqual(len(self.queue), 5)
 
     def test_peek_no_values(self):
-        self.assertEqual(self.queue.peek(5), [])
+        self.assertEqual(self.queue.peek(items=5), [])
         self.assertEqual(self.queue.peek(), None)
 
     def test_big_file_1(self):
@@ -227,7 +227,7 @@ def test_usage(self):
         self.queue.push(['a', 'b', 'c'])
 
         self.assertEqual(self.queue.peek(), 1)
-        self.assertEqual(self.queue.peek(4), [1, 2, 3, 'a'])
+        self.assertEqual(self.queue.peek(items=4), [1, 2, 3, 'a'])
         self.assertEqual(len(self.queue), 6)
 
         self.queue.push('foobar')
@@ -245,7 +245,7 @@ def test_delete(self):
 
         self.queue.delete(2)
         self.assertEqual(len(self.queue), 2)
-        self.assertEqual(self.queue.peek(2), [7, 11])
+        self.assertEqual(self.queue.peek(items=2), [7, 11])
         self.assertEqual(self.queue.get(items=2), [7, 11])
 
         self.queue.push(2)

From 34b146d5af4e61ff68323445354e6f3e5f61e6aa Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 17 Nov 2016 23:24:59 +0100
Subject: [PATCH 10/26] implement queue.put

---
 persistent_queue/persistent_queue.py | 83 ++++++++++++++--------------
 tests/test_persistent_queue.py       | 82 +++++++++++++--------------
 2 files changed, 81 insertions(+), 84 deletions(-)

diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py
index ae94adc..ec91e94 100644
--- a/persistent_queue/persistent_queue.py
+++ b/persistent_queue/persistent_queue.py
@@ -37,7 +37,7 @@ def __init__(self, filename, dumps=pickle.dumps, loads=pickle.loads, flush_limit
         self._file = self._open_file()
         self._file_lock = threading.RLock()
         self._pop_lock = threading.RLock()
-        self._pushed_event = threading.Event()
+        self._put_event = threading.Event()
 
         self._all_tasks_done = threading.Condition()
         self._unfinished_tasks = 0
@@ -105,8 +105,8 @@ def read_data():
 
         if blocking:
             while self._length < items:
-                self._pushed_event.wait()
-                self._pushed_event.clear()
+                self._put_event.wait()
+                self._put_event.clear()
 
         with self._file_lock:
             self._file.seek(self._get_queue_top(), 0)  # Beginning of data
@@ -129,42 +129,6 @@ def read_data():
 
 
 
-    def push(self, items):
-        """
-        Add items to the queue.
-
-        items: single element, or list
-        """
-        def write_data(item):
-            data = self.dumps(item)
-            self._file.write(struct.pack(LENGTH_STRUCT, len(data)))
-            self._file.write(data)
-            self._file.flush()  # Probably not necessary since buffering=0
-            os.fsync(self._file.fileno())
-
-        if not isinstance(items, list):
-            items = [items]
-
-        _LOGGER.debug("Pushing %s items", len(items))
-
-        # Ignore requests for adding zero items
-        if len(items) == 0:
-            _LOGGER.debug("Pushing zero items, ignoring request")
-            return
-
-        with self._file_lock:
-            self._file.seek(0, 2)  # Go to end of file
-
-            for i in items:
-                write_data(i)
-
-            self._update_length(self._length + len(items))
-            self._unfinished_tasks += 1
-
-        self._pushed_event.set()
-        _LOGGER.debug("Done pushing data")
-
-
 
 
     def qsize(self):
@@ -199,10 +163,16 @@ def full(self):
         """
         return False
 
-    def put(self, item, block=True, timeout=None):
+    def put(self, items, block=True, timeout=None):
         """
         Provides compatibility with stdlib Queue objects.
-        The arguments block and timeout are ignored.
+        When this function returns, all items are guaranteed to be persisted
+        into the file and the underlying storage.
+        This queue will never raise the Full exception.
+
+        items: single object, or a list of objects
+        block: this arguments is ignored.
+        timeout: this arguments is ignored.
 
         Put item into the queue. If optional args block is true and timeout is
         None (the default), block if necessary until a free slot is available.
@@ -212,7 +182,34 @@ def put(self, item, block=True, timeout=None):
         slot is immediately available, else raise the Full exception (timeout
         is ignored in that case).
         """
-        self.push(item)
+        def write_data(item):
+            data = self.dumps(item)
+            self._file.write(struct.pack(LENGTH_STRUCT, len(data)))
+            self._file.write(data)
+            self._file.flush()  # Probably not necessary since buffering=0
+            os.fsync(self._file.fileno())
+
+        if not isinstance(items, list):
+            items = [items]
+
+        _LOGGER.debug("Putting %s items", len(items))
+
+        # Ignore requests for adding zero items
+        if len(items) == 0:
+            _LOGGER.debug("Putting zero items, ignoring request")
+            return
+
+        with self._file_lock:
+            self._file.seek(0, 2)  # Go to end of file
+
+            for i in items:
+                write_data(i)
+
+            self._update_length(self._length + len(items))
+            self._unfinished_tasks += 1
+
+        self._put_event.set()
+        _LOGGER.debug("Done putting data")
 
     def put_nowait(self, item):
         """
@@ -346,7 +343,7 @@ def copy(self, new_filename):
 
     def flush(self):
         """
-        Removes elements that have been deleted or popped from the queue.
+        Removes elements that have been deleted or gotten from the queue.
         """
         _LOGGER.debug("Flushing the queue")
 
diff --git a/tests/test_persistent_queue.py b/tests/test_persistent_queue.py
index a142391..88ad9d1 100644
--- a/tests/test_persistent_queue.py
+++ b/tests/test_persistent_queue.py
@@ -23,46 +23,46 @@ def test_count(self):
         self.assertEqual(len(self.queue), 0)
         self.assertEqual(self.queue.qsize(), 0)
 
-        self.queue.push(1)
+        self.queue.put(1)
         self.assertEqual(len(self.queue), 1)
         self.assertEqual(self.queue.qsize(), 1)
 
-        self.queue.push(2)
+        self.queue.put(2)
         self.assertEqual(len(self.queue), 2)
         self.assertEqual(self.queue.qsize(), 2)
 
         for i in range(100 + 1):
-            self.queue.push(i)
+            self.queue.put(i)
 
         self.assertEqual(len(self.queue), 103)
         self.assertEqual(self.queue.qsize(), 103)
 
     def test_clear(self):
-        self.queue.push(5)
-        self.queue.push(50)
+        self.queue.put(5)
+        self.queue.put(50)
 
         self.assertEqual(self.queue.peek(items=2), [5, 50])
         self.assertEqual(len(self.queue), 2)
         self.queue.clear()
         self.assertEqual(len(self.queue), 0)
 
-    def test_push(self):
-        self.queue.push(5)
+    def test_put(self):
+        self.queue.put(5)
         self.assertEqual(self.queue.peek(items=1), 5)
 
-        self.queue.push([10, 15, 20])
+        self.queue.put([10, 15, 20])
         self.assertEqual(self.queue.peek(items=4), [5, 10, 15, 20])
 
         data = {"a": 1, "b": 2, "c": [1, 2, 3]}
-        self.queue.push(data)
+        self.queue.put(data)
         self.assertEqual(self.queue.peek(items=5), [5, 10, 15, 20, data])
 
-        self.queue.push([])
+        self.queue.put([])
         self.assertEqual(self.queue.peek(items=5), [5, 10, 15, 20, data])
 
     def test_get(self):
-        self.queue.push('a')
-        self.queue.push('b')
+        self.queue.put('a')
+        self.queue.put('b')
         self.assertEqual(len(self.queue), 2)
 
         self.assertEqual(self.queue.get(), 'a')
@@ -71,10 +71,10 @@ def test_get(self):
         self.assertEqual(self.queue.get(items=1), 'b')
         self.assertEqual(len(self.queue), 0)
 
-        self.queue.push('a')
-        self.queue.push('b')
-        self.queue.push('c')
-        self.queue.push('d')
+        self.queue.put('a')
+        self.queue.put('b')
+        self.queue.put('c')
+        self.queue.put('d')
         self.assertEqual(len(self.queue), 4)
 
         self.assertEqual(self.queue.get(items=3), ['a', 'b', 'c'])
@@ -86,7 +86,7 @@ def test_get(self):
         self.assertEqual(self.queue.get(block=False, items=100), [])
         self.assertEqual(len(self.queue), 0)
 
-        self.queue.push('d')
+        self.queue.put('d')
         self.assertEqual(self.queue.get(items=0), [])
         self.assertEqual(len(self.queue), 1)
 
@@ -96,7 +96,7 @@ def func():
             nonlocal done
             time.sleep(1)
             done = True
-            self.queue.push(5)
+            self.queue.put(5)
 
         t = threading.Thread(target=func)
         t.start()
@@ -114,9 +114,9 @@ def test_get_non_blocking_no_values(self):
             self.queue.get(block=False)
 
     def test_peek(self):
-        self.queue.push(1)
-        self.queue.push(2)
-        self.queue.push("test")
+        self.queue.put(1)
+        self.queue.put(2)
+        self.queue.put("test")
 
         self.assertEqual(self.queue.peek(), 1)
         self.assertEqual(self.queue.peek(items=1), 1)
@@ -127,7 +127,7 @@ def test_peek(self):
 
         self.queue.clear()
 
-        self.queue.push(1)
+        self.queue.put(1)
         self.assertEqual(len(self.queue), 1)
         self.assertEqual(self.queue.peek(), 1)
         self.assertEqual(self.queue.peek(items=1), 1)
@@ -141,7 +141,7 @@ def func():
             nonlocal done
             time.sleep(1)
             done = True
-            self.queue.push(5)
+            self.queue.put(5)
 
         t = threading.Thread(target=func)
         t.start()
@@ -160,7 +160,7 @@ def func():
             nonlocal done_pushing
             for i in range(5):
                 time.sleep(.1)
-                self.queue.push(i)
+                self.queue.put(i)
                 self.assertFalse(done_peeking)
             done_pushing = True
 
@@ -181,7 +181,7 @@ def test_big_file_1(self):
         data = {"a": list(range(500))}
 
         for i in range(1000):
-            self.queue.push(data)
+            self.queue.put(data)
 
         self.assertEqual(len(self.queue), 1000)
 
@@ -195,7 +195,7 @@ def test_big_file_2(self):
         data = {"a": list(range(500))}
 
         for i in range(1000):
-            self.queue.push(data)
+            self.queue.put(data)
 
         self.assertEqual(self.queue.get(items=995), [data for i in range(995)])
         self.queue.flush()
@@ -206,7 +206,7 @@ def test_big_file_2(self):
 
     def test_copy(self):
         new_queue_name = 'another_queue'
-        self.queue.push([5, 4, 3, 2, 1])
+        self.queue.put([5, 4, 3, 2, 1])
         self.assertEqual(len(self.queue), 5)
         self.assertEqual(self.queue.get(), 5)
 
@@ -221,26 +221,26 @@ def test_copy(self):
         os.remove(new_queue_name)
 
     def test_usage(self):
-        self.queue.push(1)
-        self.queue.push(2)
-        self.queue.push(3)
-        self.queue.push(['a', 'b', 'c'])
+        self.queue.put(1)
+        self.queue.put(2)
+        self.queue.put(3)
+        self.queue.put(['a', 'b', 'c'])
 
         self.assertEqual(self.queue.peek(), 1)
         self.assertEqual(self.queue.peek(items=4), [1, 2, 3, 'a'])
         self.assertEqual(len(self.queue), 6)
 
-        self.queue.push('foobar')
+        self.queue.put('foobar')
 
         self.assertEqual(self.queue.get(), 1)
         self.assertEqual(len(self.queue), 6)
         self.assertEqual(self.queue.get(items=6), [2, 3, 'a', 'b', 'c', 'foobar'])
 
     def test_delete(self):
-        self.queue.push(2)
-        self.queue.push(3)
-        self.queue.push(7)
-        self.queue.push(11)
+        self.queue.put(2)
+        self.queue.put(3)
+        self.queue.put(7)
+        self.queue.put(11)
         self.assertEqual(len(self.queue), 4)
 
         self.queue.delete(2)
@@ -248,11 +248,11 @@ def test_delete(self):
         self.assertEqual(self.queue.peek(items=2), [7, 11])
         self.assertEqual(self.queue.get(items=2), [7, 11])
 
-        self.queue.push(2)
+        self.queue.put(2)
         self.queue.delete(1000)
         self.assertEqual(len(self.queue), 0)
 
-        self.queue.push(2)
+        self.queue.put(2)
         self.queue.delete(0)
         self.assertEqual(len(self.queue), 1)
 
@@ -271,7 +271,7 @@ def random_stuff():
                     self.queue.get(items=(random_number % 5))
                 else:
                     for i in range(random_number % 10):
-                        self.queue.push({"test": [1, 2, 3], "foo": "bar", "1": random_number})
+                        self.queue.put({"test": [1, 2, 3], "foo": "bar", "1": random_number})
 
         threads = [threading.Thread(target=random_stuff) for _ in range(10)]
 
@@ -312,7 +312,7 @@ def test_big_file_1(self):
         data = {"a": list(range(1000))}
 
         for i in range(2000):
-            self.queue.push(data)
+            self.queue.put(data)
 
         self.assertEqual(len(self.queue), 2000)
 
@@ -326,7 +326,7 @@ def test_big_file_2(self):
         data = {"a": list(range(1000))}
 
         for i in range(2000):
-            self.queue.push(data)
+            self.queue.put(data)
 
         self.assertEqual(self.queue.get(items=1995), [data for i in range(1995)])
         self.queue.flush()

From ab85bc67497b61e7f2a956da8708dd0b743fbd73 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 17 Nov 2016 23:33:58 +0100
Subject: [PATCH 11/26] adapt tests

---
 persistent_queue/persistent_queue.py |   9 ---
 tests/test_persistent_queue.py       | 110 +++++++++++++++------------
 2 files changed, 61 insertions(+), 58 deletions(-)

diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py
index ec91e94..2f1d71f 100644
--- a/persistent_queue/persistent_queue.py
+++ b/persistent_queue/persistent_queue.py
@@ -125,12 +125,6 @@ def read_data():
             _LOGGER.debug("Returning data from peek")
             return data, queue_top
 
-
-
-
-
-
-
     def qsize(self):
         """
         Provides compatibility with stdlib Queue objects.
@@ -311,9 +305,6 @@ def join(self):
             while self._unfinished_tasks:
                 self._all_tasks_done.wait()
 
-
-
-
     def peek(self, block=False, items=1):
         """
         Peeks into the queue and returns items without removing them.
diff --git a/tests/test_persistent_queue.py b/tests/test_persistent_queue.py
index 88ad9d1..ced0c99 100644
--- a/tests/test_persistent_queue.py
+++ b/tests/test_persistent_queue.py
@@ -18,11 +18,14 @@ def setUp(self):
     def tearDown(self):
         os.remove(self.filename)
 
-    def test_count(self):
+    def test_qsize(self):
 
         self.assertEqual(len(self.queue), 0)
         self.assertEqual(self.queue.qsize(), 0)
 
+        self.assertTrue(self.queue.empty())
+        self.assertFalse(self.queue.full())
+
         self.queue.put(1)
         self.assertEqual(len(self.queue), 1)
         self.assertEqual(self.queue.qsize(), 1)
@@ -37,19 +40,16 @@ def test_count(self):
         self.assertEqual(len(self.queue), 103)
         self.assertEqual(self.queue.qsize(), 103)
 
-    def test_clear(self):
-        self.queue.put(5)
-        self.queue.put(50)
-
-        self.assertEqual(self.queue.peek(items=2), [5, 50])
-        self.assertEqual(len(self.queue), 2)
-        self.queue.clear()
-        self.assertEqual(len(self.queue), 0)
+        self.assertFalse(self.queue.empty())
+        self.assertFalse(self.queue.full())
 
     def test_put(self):
         self.queue.put(5)
         self.assertEqual(self.queue.peek(items=1), 5)
 
+        self.queue.put_nowait(5)
+        self.assertEqual(self.queue.get(), 5)
+
         self.queue.put([10, 15, 20])
         self.assertEqual(self.queue.peek(items=4), [5, 10, 15, 20])
 
@@ -113,6 +113,9 @@ def test_get_non_blocking_no_values(self):
         with self.assertRaises(queue.Empty):
             self.queue.get(block=False)
 
+        with self.assertRaises(queue.Empty):
+            self.queue.get_nowait()
+
     def test_peek(self):
         self.queue.put(1)
         self.queue.put(2)
@@ -177,6 +180,55 @@ def test_peek_no_values(self):
         self.assertEqual(self.queue.peek(items=5), [])
         self.assertEqual(self.queue.peek(), None)
 
+    def test_clear(self):
+        self.queue.put(5)
+        self.queue.put(50)
+
+        self.assertEqual(self.queue.peek(items=2), [5, 50])
+        self.assertEqual(len(self.queue), 2)
+        self.queue.clear()
+        self.assertEqual(len(self.queue), 0)
+
+    def test_copy(self):
+        new_queue_name = 'another_queue'
+        self.queue.put([5, 4, 3, 2, 1])
+        self.assertEqual(len(self.queue), 5)
+        self.assertEqual(self.queue.get(), 5)
+
+        new_queue = self.queue.copy(new_queue_name)
+
+        self.assertEqual(len(self.queue), len(new_queue))
+        self.assertEqual(self.queue.get(), new_queue.get())
+        self.assertEqual(self.queue.get(), new_queue.get())
+        self.assertEqual(self.queue.get(), new_queue.get())
+        self.assertEqual(self.queue.get(), new_queue.get())
+
+        os.remove(new_queue_name)
+
+    def test_delete(self):
+        self.queue.put(2)
+        self.queue.put(3)
+        self.queue.put(7)
+        self.queue.put(11)
+        self.assertEqual(len(self.queue), 4)
+
+        self.queue.delete(2)
+        self.assertEqual(len(self.queue), 2)
+        self.assertEqual(self.queue.peek(items=2), [7, 11])
+        self.assertEqual(self.queue.get(items=2), [7, 11])
+
+        self.queue.put(2)
+        self.queue.delete(1000)
+        self.assertEqual(len(self.queue), 0)
+
+        self.queue.put(2)
+        self.queue.delete(0)
+        self.assertEqual(len(self.queue), 1)
+
+    def test_delete_no_values(self):
+        self.queue.delete()
+        self.queue.delete(100)
+
     def test_big_file_1(self):
         data = {"a": list(range(500))}
 
@@ -204,22 +256,6 @@ def test_big_file_2(self):
         import time
         time.sleep(1)
 
-    def test_copy(self):
-        new_queue_name = 'another_queue'
-        self.queue.put([5, 4, 3, 2, 1])
-        self.assertEqual(len(self.queue), 5)
-        self.assertEqual(self.queue.get(), 5)
-
-        new_queue = self.queue.copy(new_queue_name)
-
-        self.assertEqual(len(self.queue), len(new_queue))
-        self.assertEqual(self.queue.get(), new_queue.get())
-        self.assertEqual(self.queue.get(), new_queue.get())
-        self.assertEqual(self.queue.get(), new_queue.get())
-        self.assertEqual(self.queue.get(), new_queue.get())
-
-        os.remove(new_queue_name)
-
     def test_usage(self):
         self.queue.put(1)
         self.queue.put(2)
@@ -236,30 +272,6 @@ def test_usage(self):
         self.assertEqual(len(self.queue), 6)
         self.assertEqual(self.queue.get(items=6), [2, 3, 'a', 'b', 'c', 'foobar'])
 
-    def test_delete(self):
-        self.queue.put(2)
-        self.queue.put(3)
-        self.queue.put(7)
-        self.queue.put(11)
-        self.assertEqual(len(self.queue), 4)
-
-        self.queue.delete(2)
-        self.assertEqual(len(self.queue), 2)
-        self.assertEqual(self.queue.peek(items=2), [7, 11])
-        self.assertEqual(self.queue.get(items=2), [7, 11])
-
-        self.queue.put(2)
-        self.queue.delete(1000)
-        self.assertEqual(len(self.queue), 0)
-
-        self.queue.put(2)
-        self.queue.delete(0)
-        self.assertEqual(len(self.queue), 1)
-
-    def test_delete_no_values(self):
-        self.queue.delete()
-        self.queue.delete(100)
-
     def test_threads(self):
         def random_stuff():
             for i in range(100):

From 56db7c572f571de4034c3a97ad71cdc82907bcec Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 17 Nov 2016 23:39:21 +0100
Subject: [PATCH 12/26] adapt travis changes

---
 .travis.yml | 7 +++++++
 tox.ini     | 2 +-
 2 files changed, 8 insertions(+), 1 deletion(-)

diff --git a/.travis.yml b/.travis.yml
index 43b5678..ac70718 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -6,10 +6,17 @@ matrix:
   include:
     - python: 3.5
       env: TOXENV=lint
+    - python: 2.7
+      env: TOXENV=py27
     - python: 3.4
       env: env: TOXENV=py34
     - python: 3.5
       env: env: TOXENV=py35
+    - python: "3.6-dev"
+      env: env: TOXENV=py36
+  allow_failures:
+    - python: "3.6-dev"
+      env: env: TOXENV=py36
 install:
   - "pip install tox"
 script:
diff --git a/tox.ini b/tox.ini
index 708565f..00e238a 100644
--- a/tox.ini
+++ b/tox.ini
@@ -1,5 +1,5 @@
 [tox]
-envlist = py34, py35, lint
+envlist = py27, py34, py35, py36, lint
 skipsdist = True
 toxworkdir = {env:TOX_WORK_DIR:.tox}
 

From da7586ac866333188de8b74db0a26249134e1565 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 17 Nov 2016 23:41:53 +0100
Subject: [PATCH 13/26] fix os.replace on py27

---
 persistent_queue/persistent_queue.py | 3 ++-
 setup.py                             | 3 ++-
 2 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py
index 2f1d71f..9ab5405 100644
--- a/persistent_queue/persistent_queue.py
+++ b/persistent_queue/persistent_queue.py
@@ -386,7 +386,8 @@ def flush(self):
             # the danger zone.
 
             _LOGGER.debug("Replacing old file with new file")
-            os.replace(temp_filename, self.filename)
+            os.remove(self.filename)
+            os.rename(temp_filename, self.filename)
             self._file = self._open_file()
 
             _LOGGER.debug("Finished flushing the queue")
diff --git a/setup.py b/setup.py
index 9158162..f3e3244 100644
--- a/setup.py
+++ b/setup.py
@@ -25,8 +25,9 @@
         "License :: OSI Approved :: MIT License",
         "Development Status :: 5 - Production/Stable",
         "Programming Language :: Python",
+        "Programming Language :: Python :: 2",
+        "Programming Language :: Python :: 2.7",
         "Programming Language :: Python :: 3",
-        "Programming Language :: Python :: 3 :: Only",
         "Programming Language :: Python :: 3.4",
         "Programming Language :: Python :: 3.5",
         "Topic :: Software Development :: Libraries :: Python Modules",

From 2e2c70100d2f5bda06874a713796434673a47dc5 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 17 Nov 2016 23:57:36 +0100
Subject: [PATCH 14/26] add py27 compatibility

---
 persistent_queue/__init__.py         |  2 ++
 persistent_queue/persistent_queue.py | 16 +++++++-----
 tests/test_persistent_queue.py       | 37 ++++++++++++++--------------
 3 files changed, 31 insertions(+), 24 deletions(-)

diff --git a/persistent_queue/__init__.py b/persistent_queue/__init__.py
index a79ae19..3a08262 100644
--- a/persistent_queue/__init__.py
+++ b/persistent_queue/__init__.py
@@ -1,3 +1,5 @@
+from __future__ import absolute_import
+
 from .persistent_queue import PersistentQueue
 
 __all__ = [
diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py
index 9ab5405..cce7f07 100644
--- a/persistent_queue/persistent_queue.py
+++ b/persistent_queue/persistent_queue.py
@@ -10,7 +10,11 @@
 import struct
 import threading
 import uuid
-import queue
+
+try:
+    import queue
+except ImportError:
+    import Queue as queue
 
 LENGTH_STRUCT = 'I'
 HEADER_STRUCT = 'II'
@@ -129,7 +133,7 @@ def qsize(self):
         """
         Provides compatibility with stdlib Queue objects.
 
-        Return the approximate size of the queue. Note, qsize() > 0 doesn’t
+        Return the approximate size of the queue. Note, qsize() > 0 doesn't
         guarantee that a subsequent get() will not block, nor will qsize() <
         maxsize guarantee that put() will not block.
         """
@@ -140,8 +144,8 @@ def empty(self):
         Provides compatibility with stdlib Queue objects.
 
         Return True if the queue is empty, False otherwise. If empty() returns
-        True it doesn’t guarantee that a subsequent call to put() will not
-        block. Similarly, if empty() returns False it doesn’t guarantee that a
+        True it doesn't guarantee that a subsequent call to put() will not
+        block. Similarly, if empty() returns False it doesn't guarantee that a
         subsequent call to get() will not block.
         """
         return self._length == 0
@@ -151,8 +155,8 @@ def full(self):
         Provides compatibility with stdlib Queue objects.
 
         Return True if the queue is full, False otherwise. If full() returns
-        True it doesn’t guarantee that a subsequent call to get() will not
-        block. Similarly, if full() returns False it doesn’t guarantee that a
+        True it doesn't guarantee that a subsequent call to get() will not
+        block. Similarly, if full() returns False it doesn't guarantee that a
         subsequent call to put() will not block.
         """
         return False
diff --git a/tests/test_persistent_queue.py b/tests/test_persistent_queue.py
index ced0c99..a9ab00e 100644
--- a/tests/test_persistent_queue.py
+++ b/tests/test_persistent_queue.py
@@ -4,7 +4,11 @@
 import time
 import unittest
 import uuid
-import queue
+
+try:
+    import queue
+except ImportError:
+    import Queue as queue
 
 from persistent_queue import PersistentQueue
 
@@ -91,19 +95,18 @@ def test_get(self):
         self.assertEqual(len(self.queue), 1)
 
     def test_get_blocking(self):
-        done = False
+        done = [False]
         def func():
-            nonlocal done
             time.sleep(1)
-            done = True
+            done[0] = True
             self.queue.put(5)
 
         t = threading.Thread(target=func)
         t.start()
 
-        self.assertFalse(done)
+        self.assertFalse(done[0])
         data = self.queue.get()
-        self.assertTrue(done)
+        self.assertTrue(done[0])
         self.assertEqual(data, 5)
         self.assertEqual(len(self.queue), 0)
 
@@ -139,40 +142,38 @@ def test_peek(self):
         self.assertEqual(self.queue.peek(items=0), [])
 
     def test_peek_blocking(self):
-        done = False
+        done = [False]
         def func():
-            nonlocal done
             time.sleep(1)
-            done = True
+            done[0] = True
             self.queue.put(5)
 
         t = threading.Thread(target=func)
         t.start()
 
-        self.assertFalse(done)
+        self.assertFalse(done[0])
         data = self.queue.peek(block=True)
-        self.assertTrue(done)
+        self.assertTrue(done[0])
         self.assertEqual(data, 5)
         self.assertEqual(len(self.queue), 1)
 
     def test_peek_blocking_list(self):
-        done_pushing = False
-        done_peeking = False
+        done_pushing = [False]
+        done_peeking = [False]
 
         def func():
-            nonlocal done_pushing
             for i in range(5):
                 time.sleep(.1)
                 self.queue.put(i)
-                self.assertFalse(done_peeking)
-            done_pushing = True
+                self.assertFalse(done_peeking[0])
+            done_pushing[0] = True
 
         t = threading.Thread(target=func)
         t.start()
 
         data = self.queue.peek(items=5, block=True)
-        done_peeking = True
-        self.assertTrue(done_pushing)
+        done_peeking[0] = True
+        self.assertTrue(done_pushing[0])
         self.assertEqual(data, [0, 1, 2, 3, 4])
         self.assertEqual(len(self.queue), 5)
 

From 3b9842cfbad073a2b8550ce23f74cc248550f4a7 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Fri, 18 Nov 2016 00:24:37 +0100
Subject: [PATCH 15/26] fix travis

---
 .travis.yml | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/.travis.yml b/.travis.yml
index ac70718..5c60502 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -9,14 +9,14 @@ matrix:
     - python: 2.7
       env: TOXENV=py27
     - python: 3.4
-      env: env: TOXENV=py34
+      env: TOXENV=py34
     - python: 3.5
-      env: env: TOXENV=py35
+      env: TOXENV=py35
     - python: "3.6-dev"
-      env: env: TOXENV=py36
+      env: TOXENV=py36
   allow_failures:
     - python: "3.6-dev"
-      env: env: TOXENV=py36
+      env: TOXENV=py36
 install:
   - "pip install tox"
 script:

From 30c1c01f386dba54a3d08369b7e286ddfe042f8d Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Fri, 18 Nov 2016 10:53:28 +0100
Subject: [PATCH 16/26] implement put timeout

---
 persistent_queue/persistent_queue.py | 72 +++++++++++++++++++---------
 tests/test_persistent_queue.py       | 25 ++++++++--
 2 files changed, 72 insertions(+), 25 deletions(-)

diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py
index cce7f07..9b18fb0 100644
--- a/persistent_queue/persistent_queue.py
+++ b/persistent_queue/persistent_queue.py
@@ -9,11 +9,12 @@
 import shutil
 import struct
 import threading
+import time
 import uuid
 
 try:
     import queue
-except ImportError:
+except ImportError:  # pragma: no cover
     import Queue as queue
 
 LENGTH_STRUCT = 'I'
@@ -24,15 +25,22 @@
 
 
 class PersistentQueue:
-    def __init__(self, filename, dumps=pickle.dumps, loads=pickle.loads, flush_limit=1048576):
+    def __init__(self, maxsize=0, filename=None, dumps=pickle.dumps, loads=pickle.loads, flush_limit=1048576):
         """
         Creates a new PersistentQueue object and underlying file.
 
+        maxsize: upperbound limit of items that can be placed in the queue.
         filename: must be a full path to the new file.
         dumps: the function called for persisting queue items to the file.
         loads: the function called for loading queue items from the file.
         flush_limit: below this filesize, flush() is a no-op.
         """
+        if filename is None:
+            raise ValueError("filename must be set, e.g. '/my/path/important.queue'")
+        if maxsize < 0:
+            maxsize = 0
+
+        self.maxsize = maxsize
         self.filename = os.path.abspath(filename)
         self.dumps = dumps
         self.loads = loads
@@ -40,7 +48,9 @@ def __init__(self, filename, dumps=pickle.dumps, loads=pickle.loads, flush_limit
 
         self._file = self._open_file()
         self._file_lock = threading.RLock()
-        self._pop_lock = threading.RLock()
+        self._get_lock = threading.RLock()
+        self._get_event = threading.Event()
+        self._put_lock = threading.RLock()
         self._put_event = threading.Event()
 
         self._all_tasks_done = threading.Condition()
@@ -90,7 +100,7 @@ def _set_queue_top(self, top):
 
         self._file.seek(current_pos, 0)
 
-    def _peek(self, items=1, blocking=False):
+    def _peek(self, items=1, block=False):
         """
         Returns a certain amount of items from the queue. If items is greater
         than one, a list is returned.
@@ -107,7 +117,7 @@ def read_data():
             _LOGGER.debug("Returning empty list")
             return [], self._file.tell()
 
-        if blocking:
+        if block:
             while self._length < items:
                 self._put_event.wait()
                 self._put_event.clear()
@@ -166,11 +176,8 @@ def put(self, items, block=True, timeout=None):
         Provides compatibility with stdlib Queue objects.
         When this function returns, all items are guaranteed to be persisted
         into the file and the underlying storage.
-        This queue will never raise the Full exception.
 
         items: single object, or a list of objects
-        block: this arguments is ignored.
-        timeout: this arguments is ignored.
 
         Put item into the queue. If optional args block is true and timeout is
         None (the default), block if necessary until a free slot is available.
@@ -197,17 +204,33 @@ def write_data(item):
             _LOGGER.debug("Putting zero items, ignoring request")
             return
 
-        with self._file_lock:
-            self._file.seek(0, 2)  # Go to end of file
+        with self._put_lock:
+            if self.maxsize > 0:
+                if block:
+                    if timeout is not None:
+                        target = time.time() + timeout
+                    while self._length + len(items) > self.maxsize:
+                        if self._get_event.wait(timeout) is not True:
+                            if timeout is not None:
+                                timeout = target - time.time()
+                                if timeout <= 0:
+                                    raise queue.Full
+                        self._get_event.clear()
+                else:
+                    if self._length + len(items) > self.maxsize:
+                        raise queue.Full
+
+            with self._file_lock:
+                self._file.seek(0, 2)  # Go to end of file
 
-            for i in items:
-                write_data(i)
+                for i in items:
+                    write_data(i)
 
-            self._update_length(self._length + len(items))
-            self._unfinished_tasks += 1
+                self._update_length(self._length + len(items))
+                self._unfinished_tasks += 1
 
-        self._put_event.set()
-        _LOGGER.debug("Done putting data")
+            self._put_event.set()
+            _LOGGER.debug("Done putting data")
 
     def put_nowait(self, item):
         """
@@ -241,7 +264,7 @@ def get(self, block=True, timeout=None, items=1):
             _LOGGER.debug("Returning empty list")
             return []
 
-        with self._pop_lock:
+        with self._get_lock:
             data, queue_top = self._peek(items, block)
 
             with self._file_lock:
@@ -255,6 +278,7 @@ def get(self, block=True, timeout=None, items=1):
                 elif data is None:
                     raise queue.Empty
 
+                self._get_event.set()
                 _LOGGER.debug("Returning data from get")
                 return data
 
@@ -313,7 +337,7 @@ def peek(self, block=False, items=1):
         """
         Peeks into the queue and returns items without removing them.
         """
-        with self._pop_lock:
+        with self._get_lock:
             return self._peek(items, block)[0]
 
     def clear(self):
@@ -321,7 +345,7 @@ def clear(self):
         Removes all elements from queue, by truncating the file and reloading.
         """
         _LOGGER.debug("Clearing the queue")
-        with self._file_lock, self._pop_lock:
+        with self._file_lock, self._get_lock:
             self._file.close()
             self._file = self._open_file(mode='w+b')
             self._length = 0
@@ -334,7 +358,11 @@ def copy(self, new_filename):
         new_filename: must be a full path to the new file.
         """
         shutil.copy2(self.filename, new_filename)
-        return PersistentQueue(filename=new_filename, flush_limit=self.flush_limit)
+        return PersistentQueue(maxsize=self.maxsize,
+                               filename=new_filename,
+                               dumps=self.dumps,
+                               loads=self.loads,
+                               flush_limit=self.flush_limit)
 
     def flush(self):
         """
@@ -356,7 +384,7 @@ def flush(self):
         new_file = open(temp_filename, mode='w+b', buffering=0)
 
         # From this point on, the file can not change
-        with self._file_lock, self._pop_lock:
+        with self._file_lock, self._get_lock:
             # Make sure everything is to disk
             self._file.flush()
             os.fsync(self._file.fileno())
@@ -413,7 +441,7 @@ def read_length():
             _LOGGER.debug("Ignoring request to delete")
             return
 
-        with self._file_lock, self._pop_lock:
+        with self._file_lock, self._get_lock:
             self._file.seek(self._get_queue_top(), 0)  # Beginning of data
             total_items = self._length if items > self._length else items
 
diff --git a/tests/test_persistent_queue.py b/tests/test_persistent_queue.py
index a9ab00e..0541fe6 100644
--- a/tests/test_persistent_queue.py
+++ b/tests/test_persistent_queue.py
@@ -17,11 +17,21 @@ class TestPersistentQueue(unittest.TestCase):
     def setUp(self):
         random = str(uuid.uuid4()).replace('-', '')
         self.filename = '{}_{}.queue'.format(self.id(), random)
-        self.queue = PersistentQueue(self.filename)
+        self.queue = PersistentQueue(filename=self.filename)
 
     def tearDown(self):
         os.remove(self.filename)
 
+    def test_simple(self):
+        with self.assertRaises(ValueError):
+            q = PersistentQueue()
+
+        q = PersistentQueue(filename='foobar')
+        self.assertEqual(q.maxsize, 0)
+
+        q = PersistentQueue(maxsize=-42, filename='foobar')
+        self.assertEqual(q.maxsize, 0)
+
     def test_qsize(self):
 
         self.assertEqual(len(self.queue), 0)
@@ -64,6 +74,13 @@ def test_put(self):
         self.queue.put([])
         self.assertEqual(self.queue.peek(items=5), [5, 10, 15, 20, data])
 
+        self.queue.maxsize = 4
+        with self.assertRaises(queue.Full):
+            self.queue.put('full', timeout=1)
+
+        with self.assertRaises(queue.Full):
+            self.queue.put('full', block=False)
+
     def test_get(self):
         self.queue.put('a')
         self.queue.put('b')
@@ -305,7 +322,8 @@ def setUp(self):
 
         random = str(uuid.uuid4()).replace('-', '')
         self.filename = '{}_{}'.format(self.id(), random)
-        self.queue = PersistentQueue(self.filename, loads=dill.loads,
+        self.queue = PersistentQueue(filename=self.filename,
+                                     loads=dill.loads,
                                      dumps=dill.dumps)
 
 
@@ -315,7 +333,8 @@ def setUp(self):
 
         random = str(uuid.uuid4()).replace('-', '')
         self.filename = '{}_{}'.format(self.id(), random)
-        self.queue = PersistentQueue(self.filename, loads=bson.loads,
+        self.queue = PersistentQueue(filename=self.filename,
+                                     loads=bson.loads,
                                      dumps=bson.dumps)
 
     def tearDown(self):

From 92ac3bbdc852537568e98b0802d171ea03ee9ab0 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Fri, 18 Nov 2016 10:55:24 +0100
Subject: [PATCH 17/26] test cleanup

---
 tests/test_persistent_queue.py | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/tests/test_persistent_queue.py b/tests/test_persistent_queue.py
index 0541fe6..ba1fc06 100644
--- a/tests/test_persistent_queue.py
+++ b/tests/test_persistent_queue.py
@@ -23,15 +23,20 @@ def tearDown(self):
         os.remove(self.filename)
 
     def test_simple(self):
+        random = str(uuid.uuid4()).replace('-', '')
+        filename = '{}_{}.queue'.format(self.id(), random)
+
         with self.assertRaises(ValueError):
             q = PersistentQueue()
 
-        q = PersistentQueue(filename='foobar')
+        q = PersistentQueue(filename=filename)
         self.assertEqual(q.maxsize, 0)
 
-        q = PersistentQueue(maxsize=-42, filename='foobar')
+        q = PersistentQueue(maxsize=-42, filename=filename)
         self.assertEqual(q.maxsize, 0)
 
+        os.remove(filename)
+
     def test_qsize(self):
 
         self.assertEqual(len(self.queue), 0)

From 9365a1f829948f83a639b2b91f864ea776ca8b0b Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Fri, 18 Nov 2016 11:31:13 +0100
Subject: [PATCH 18/26] implement get timeout

---
 persistent_queue/persistent_queue.py | 26 +++++++++++++++-----------
 tests/test_persistent_queue.py       |  3 +++
 2 files changed, 18 insertions(+), 11 deletions(-)

diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py
index 9b18fb0..a266b24 100644
--- a/persistent_queue/persistent_queue.py
+++ b/persistent_queue/persistent_queue.py
@@ -100,7 +100,7 @@ def _set_queue_top(self, top):
 
         self._file.seek(current_pos, 0)
 
-    def _peek(self, items=1, block=False):
+    def _peek(self, block, timeout, items, partial=False):
         """
         Returns a certain amount of items from the queue. If items is greater
         than one, a list is returned.
@@ -118,9 +118,18 @@ def read_data():
             return [], self._file.tell()
 
         if block:
+            if timeout is not None:
+                target = time.time() + timeout
             while self._length < items:
-                self._put_event.wait()
+                if self._put_event.wait(timeout) is not True:
+                    if timeout is not None:
+                        timeout = target - time.time()
+                        if timeout <= 0:
+                            raise queue.Empty
                 self._put_event.clear()
+        elif not partial and items == 1:
+            if self._length < items:
+                raise queue.Empty
 
         with self._file_lock:
             self._file.seek(self._get_queue_top(), 0)  # Beginning of data
@@ -169,7 +178,7 @@ def full(self):
         block. Similarly, if full() returns False it doesn't guarantee that a
         subsequent call to put() will not block.
         """
-        return False
+        return self.maxsize > 0 and self._length >= self.maxsize
 
     def put(self, items, block=True, timeout=None):
         """
@@ -243,7 +252,6 @@ def put_nowait(self, item):
     def get(self, block=True, timeout=None, items=1):
         """
         Provides compatibility with stdlib Queue objects.
-        timeout: this argument is ignored.
         items: number of how many items are returned. If items is greater than
         one, a list is returned.
 
@@ -257,15 +265,13 @@ def get(self, block=True, timeout=None, items=1):
         """
         _LOGGER.debug("Getting %s items", items)
 
-        # TODO: implement timeout
-
         # Ignore requests for zero items
         if items == 0:
             _LOGGER.debug("Returning empty list")
             return []
 
         with self._get_lock:
-            data, queue_top = self._peek(items, block)
+            data, queue_top = self._peek(block, timeout, items)
 
             with self._file_lock:
                 self._set_queue_top(queue_top)
@@ -275,8 +281,6 @@ def get(self, block=True, timeout=None, items=1):
                         self._update_length(self._length - len(data))
                 elif data is not None:
                     self._update_length(self._length - 1)
-                elif data is None:
-                    raise queue.Empty
 
                 self._get_event.set()
                 _LOGGER.debug("Returning data from get")
@@ -333,12 +337,12 @@ def join(self):
             while self._unfinished_tasks:
                 self._all_tasks_done.wait()
 
-    def peek(self, block=False, items=1):
+    def peek(self, block=False, timeout=None, items=1):
         """
         Peeks into the queue and returns items without removing them.
         """
         with self._get_lock:
-            return self._peek(items, block)[0]
+            return self._peek(block, timeout, items, partial=True)[0]
 
     def clear(self):
         """
diff --git a/tests/test_persistent_queue.py b/tests/test_persistent_queue.py
index ba1fc06..22d3bd4 100644
--- a/tests/test_persistent_queue.py
+++ b/tests/test_persistent_queue.py
@@ -132,6 +132,9 @@ def func():
         self.assertEqual(data, 5)
         self.assertEqual(len(self.queue), 0)
 
+        with self.assertRaises(queue.Empty):
+            self.queue.get(timeout=1)
+
     def test_get_non_blocking_no_values(self):
         self.assertEqual(self.queue.get(block=False, items=5), [])
 

From 6230352837a09c3ffba600abdd8fb8b1d73cdb1e Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Fri, 18 Nov 2016 11:49:53 +0100
Subject: [PATCH 19/26] add tests for task_done and join

---
 persistent_queue/persistent_queue.py |  6 +++---
 tests/test_persistent_queue.py       | 30 ++++++++++++++++++++++++++--
 2 files changed, 31 insertions(+), 5 deletions(-)

diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py
index a266b24..0de1e2b 100644
--- a/persistent_queue/persistent_queue.py
+++ b/persistent_queue/persistent_queue.py
@@ -236,7 +236,7 @@ def write_data(item):
                     write_data(i)
 
                 self._update_length(self._length + len(items))
-                self._unfinished_tasks += 1
+                self._unfinished_tasks += len(items)
 
             self._put_event.set()
             _LOGGER.debug("Done putting data")
@@ -297,7 +297,7 @@ def get_nowait(self):
         """
         return self.get(block=False)
 
-    def task_done(self):
+    def task_done(self, items=1):
         """
         Provides compatibility with stdlib Queue objects.
         Taken from the Python 3.5 stdlib: lib/python3.5/queue.py
@@ -314,7 +314,7 @@ def task_done(self):
         Raises a ValueError if called more times than there were items placed in the queue.
         """
         with self._all_tasks_done:
-            unfinished = self._unfinished_tasks - 1
+            unfinished = self._unfinished_tasks - items
             if unfinished <= 0:
                 if unfinished < 0:
                     raise ValueError('task_done() called too many times')
diff --git a/tests/test_persistent_queue.py b/tests/test_persistent_queue.py
index 22d3bd4..2dae179 100644
--- a/tests/test_persistent_queue.py
+++ b/tests/test_persistent_queue.py
@@ -304,9 +304,15 @@ def random_stuff():
                 random_number = random.randint(0, 1000)
 
                 if random_number % 3 == 0:
-                    self.queue.peek(random_number % 5)
+                    try:
+                        self.queue.peek(block=False, items=(random_number % 5))
+                    except queue.Empty:
+                        pass
                 elif random_number % 2 == 0:
-                    self.queue.get(items=(random_number % 5))
+                    try:
+                        self.queue.get(block=False, items=(random_number % 5))
+                    except queue.Empty:
+                        pass
                 else:
                     for i in range(random_number % 10):
                         self.queue.put({"test": [1, 2, 3], "foo": "bar", "1": random_number})
@@ -323,6 +329,26 @@ def random_stuff():
         for _ in range(len(self.queue)):
             self.queue.get()
 
+    def test_join_on_task_done(self):
+        def worker():
+            while True:
+                try:
+                    i = self.queue.get(block=False)
+                    self.queue.task_done()
+                except queue.Empty:
+                    with self.assertRaises(ValueError):
+                        # called too many times
+                        self.queue.task_done()
+                    return
+
+        self.queue.put(list(range(10)))
+
+        t = threading.Thread(target=worker)
+        t.start()
+
+        self.queue.join()
+        self.assertTrue(self.queue.empty())
+
 
 class TestPersistentQueueWithDill(TestPersistentQueue):
     def setUp(self):

From 81a3eb2aa353a4e3eccbea269c2bac67803c43af Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Fri, 18 Nov 2016 11:59:26 +0100
Subject: [PATCH 20/26] lint tests too

---
 setup.cfg                      | 1 -
 tests/test_persistent_queue.py | 4 +++-
 tox.ini                        | 2 +-
 3 files changed, 4 insertions(+), 3 deletions(-)

diff --git a/setup.cfg b/setup.cfg
index 48513d1..790df1e 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -2,7 +2,6 @@
 max-line-length = 140
 max-complexity = 25
 ignore = E251,C901,W503
-exclude = tests/*
 addons = file,open,basestring,xrange,unicode,long,cmp
 
 [tool:pytest]
diff --git a/tests/test_persistent_queue.py b/tests/test_persistent_queue.py
index 2dae179..0c3c064 100644
--- a/tests/test_persistent_queue.py
+++ b/tests/test_persistent_queue.py
@@ -118,6 +118,7 @@ def test_get(self):
 
     def test_get_blocking(self):
         done = [False]
+
         def func():
             time.sleep(1)
             done[0] = True
@@ -168,6 +169,7 @@ def test_peek(self):
 
     def test_peek_blocking(self):
         done = [False]
+
         def func():
             time.sleep(1)
             done[0] = True
@@ -333,7 +335,7 @@ def test_join_on_task_done(self):
         def worker():
             while True:
                 try:
-                    i = self.queue.get(block=False)
+                    self.queue.get(block=False)
                     self.queue.task_done()
                 except queue.Empty:
                     with self.assertRaises(ValueError):
diff --git a/tox.ini b/tox.ini
index 00e238a..e210028 100644
--- a/tox.ini
+++ b/tox.ini
@@ -18,4 +18,4 @@ commands = py.test -v --cov persistent_queue {posargs}
 deps =
   flake8==3.0.4
 commands =
-  flake8 --count persistent_queue
+  flake8 --count persistent_queue tests

From cafd4139a1c3b3be882231eff1cae7a2cf9a4376 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 1 Dec 2016 11:53:57 +0100
Subject: [PATCH 21/26] change unfinished checks

---
 persistent_queue/persistent_queue.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py
index 0de1e2b..7118518 100644
--- a/persistent_queue/persistent_queue.py
+++ b/persistent_queue/persistent_queue.py
@@ -315,9 +315,9 @@ def task_done(self, items=1):
         """
         with self._all_tasks_done:
             unfinished = self._unfinished_tasks - items
-            if unfinished <= 0:
-                if unfinished < 0:
-                    raise ValueError('task_done() called too many times')
+            if unfinished < 0:
+                raise ValueError('task_done() called too many times')
+            if unfinished == 0:
                 self._all_tasks_done.notify_all()
             self._unfinished_tasks = unfinished
 

From 87cc620a233b951359bead634184203a8b545a77 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 1 Dec 2016 21:31:05 +0100
Subject: [PATCH 22/26] change init API

---
 persistent_queue/persistent_queue.py | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py
index 7118518..74f15ac 100644
--- a/persistent_queue/persistent_queue.py
+++ b/persistent_queue/persistent_queue.py
@@ -25,18 +25,16 @@
 
 
 class PersistentQueue:
-    def __init__(self, maxsize=0, filename=None, dumps=pickle.dumps, loads=pickle.loads, flush_limit=1048576):
+    def __init__(self, filename, maxsize=0, dumps=pickle.dumps, loads=pickle.loads, flush_limit=1048576):
         """
         Creates a new PersistentQueue object and underlying file.
 
-        maxsize: upperbound limit of items that can be placed in the queue.
         filename: must be a full path to the new file.
+        maxsize: upperbound limit of items that can be placed in the queue.
         dumps: the function called for persisting queue items to the file.
         loads: the function called for loading queue items from the file.
         flush_limit: below this filesize, flush() is a no-op.
         """
-        if filename is None:
-            raise ValueError("filename must be set, e.g. '/my/path/important.queue'")
         if maxsize < 0:
             maxsize = 0
 

From 2c502079dff2475add3127da282d392956be370b Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 1 Dec 2016 21:33:48 +0100
Subject: [PATCH 23/26] change partial behaviour

---
 persistent_queue/persistent_queue.py | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py
index 74f15ac..fa51bf6 100644
--- a/persistent_queue/persistent_queue.py
+++ b/persistent_queue/persistent_queue.py
@@ -125,9 +125,8 @@ def read_data():
                         if timeout <= 0:
                             raise queue.Empty
                 self._put_event.clear()
-        elif not partial and items == 1:
-            if self._length < items:
-                raise queue.Empty
+        elif not partial and self._length < items:
+            raise queue.Empty
 
         with self._file_lock:
             self._file.seek(self._get_queue_top(), 0)  # Beginning of data

From 5d010b3bcad909737c8ad84f8cc0319d067f2691 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 1 Dec 2016 21:46:23 +0100
Subject: [PATCH 24/26] fix block+timeout behaviour

---
 persistent_queue/persistent_queue.py | 38 +++++++++++++++++++---------
 1 file changed, 26 insertions(+), 12 deletions(-)

diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py
index fa51bf6..94ebbb4 100644
--- a/persistent_queue/persistent_queue.py
+++ b/persistent_queue/persistent_queue.py
@@ -119,12 +119,19 @@ def read_data():
             if timeout is not None:
                 target = time.time() + timeout
             while self._length < items:
-                if self._put_event.wait(timeout) is not True:
-                    if timeout is not None:
-                        timeout = target - time.time()
-                        if timeout <= 0:
-                            raise queue.Empty
-                self._put_event.clear()
+                if self._put_event.wait(timeout) is False:
+                    # Nothing was added to the queue and timeout expired
+                    # This will never happen if timeout is None
+                    raise queue.Empty
+
+                # Something was added to the queue
+                # Update timeout, if necessary
+                if timeout is not None:
+                    timeout = target - time.time()
+                    if timeout <= 0:
+                        raise queue.Empty
+
+            self._put_event.clear()
         elif not partial and self._length < items:
             raise queue.Empty
 
@@ -216,12 +223,19 @@ def write_data(item):
                     if timeout is not None:
                         target = time.time() + timeout
                     while self._length + len(items) > self.maxsize:
-                        if self._get_event.wait(timeout) is not True:
-                            if timeout is not None:
-                                timeout = target - time.time()
-                                if timeout <= 0:
-                                    raise queue.Full
-                        self._get_event.clear()
+                        if self._get_event.wait(timeout) is False:
+                            # Nothing was removed from the queue and timeout expired
+                            # This will never happen if timeout is None
+                            raise queue.Full
+
+                        # Something was removed from the queue
+                        # Update timeout, if necessary
+                        if timeout is not None:
+                            timeout = target - time.time()
+                            if timeout <= 0:
+                                raise queue.Full
+
+                    self._get_event.clear()
                 else:
                     if self._length + len(items) > self.maxsize:
                         raise queue.Full

From d3b63fb27821faf0733a7a745a6ea80e9d66737a Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Thu, 1 Dec 2016 21:54:53 +0100
Subject: [PATCH 25/26] update tests and lint

---
 tests/test_persistent_queue.py | 15 +++++++--------
 1 file changed, 7 insertions(+), 8 deletions(-)

diff --git a/tests/test_persistent_queue.py b/tests/test_persistent_queue.py
index 0c3c064..a4f2eec 100644
--- a/tests/test_persistent_queue.py
+++ b/tests/test_persistent_queue.py
@@ -26,7 +26,7 @@ def test_simple(self):
         random = str(uuid.uuid4()).replace('-', '')
         filename = '{}_{}.queue'.format(self.id(), random)
 
-        with self.assertRaises(ValueError):
+        with self.assertRaises(TypeError):
             q = PersistentQueue()
 
         q = PersistentQueue(filename=filename)
@@ -106,15 +106,13 @@ def test_get(self):
         self.assertEqual(self.queue.get(items=3), ['a', 'b', 'c'])
         self.assertEqual(len(self.queue), 1)
 
-        self.assertEqual(self.queue.get(block=False, items=100), ['d'])
-        self.assertEqual(len(self.queue), 0)
-
-        self.assertEqual(self.queue.get(block=False, items=100), [])
-        self.assertEqual(len(self.queue), 0)
+        with self.assertRaises(queue.Empty):
+            self.assertEqual(self.queue.get(block=False, items=100), ['d'])
+        self.assertEqual(len(self.queue), 1)
 
         self.queue.put('d')
         self.assertEqual(self.queue.get(items=0), [])
-        self.assertEqual(len(self.queue), 1)
+        self.assertEqual(len(self.queue), 2)
 
     def test_get_blocking(self):
         done = [False]
@@ -137,7 +135,8 @@ def func():
             self.queue.get(timeout=1)
 
     def test_get_non_blocking_no_values(self):
-        self.assertEqual(self.queue.get(block=False, items=5), [])
+        with self.assertRaises(queue.Empty):
+            self.assertEqual(self.queue.get(block=False, items=5), [])
 
         with self.assertRaises(queue.Empty):
             self.queue.get(block=False)

From 1b372ff18235926b46190a42a8415bec30b81e40 Mon Sep 17 00:00:00 2001
From: Thomas Kriechbaumer <thomas@kriechbaumer.name>
Date: Fri, 2 Dec 2016 22:48:40 +0100
Subject: [PATCH 26/26] remove unreachable code

---
 persistent_queue/persistent_queue.py | 9 ++-------
 1 file changed, 2 insertions(+), 7 deletions(-)

diff --git a/persistent_queue/persistent_queue.py b/persistent_queue/persistent_queue.py
index 94ebbb4..bdeb5f2 100644
--- a/persistent_queue/persistent_queue.py
+++ b/persistent_queue/persistent_queue.py
@@ -123,15 +123,13 @@ def read_data():
                     # Nothing was added to the queue and timeout expired
                     # This will never happen if timeout is None
                     raise queue.Empty
+                self._put_event.clear()
 
                 # Something was added to the queue
                 # Update timeout, if necessary
                 if timeout is not None:
                     timeout = target - time.time()
-                    if timeout <= 0:
-                        raise queue.Empty
 
-            self._put_event.clear()
         elif not partial and self._length < items:
             raise queue.Empty
 
@@ -227,15 +225,12 @@ def write_data(item):
                             # Nothing was removed from the queue and timeout expired
                             # This will never happen if timeout is None
                             raise queue.Full
+                        self._get_event.clear()
 
                         # Something was removed from the queue
                         # Update timeout, if necessary
                         if timeout is not None:
                             timeout = target - time.time()
-                            if timeout <= 0:
-                                raise queue.Full
-
-                    self._get_event.clear()
                 else:
                     if self._length + len(items) > self.maxsize:
                         raise queue.Full