Upgrade Celery to 4.x (close snowplow#210)

Author: mhadam

setup.py

@@ -78,7 +78,7 @@
         "greenlet>=0.4.10,<1.0",
         "requests>=2.2.1,<3.0",
         "pycontracts>=1.7.6,<2.0",
-        "celery>=3.1.11,<4.0",
+        "celery>=4.0,<5.0",
         "gevent>=1.0.2,<2.0",
         "redis>=2.9.1,<3.0",
         "six>=1.9.0,<2.0"

snowplow_tracker/emitters.py

@@ -31,7 +31,6 @@
     from queue import Queue
 
 from celery import Celery
-from celery.contrib.methods import task
 import redis
 import requests
 from contracts import contract, new_contract
@@ -57,9 +56,9 @@
     import snowplow_celery_config
     app = Celery()
     app.config_from_object(snowplow_celery_config)
-
 except ImportError:
     # Otherwise configure Celery with default settings
+    snowplow_celery_config = None
     app = Celery("Snowplow", broker="redis://guest@localhost//")
 
 
@@ -172,7 +171,6 @@ def reached_limit(self):
         else:
             return self.bytes_queued >= self.byte_limit or len(self.buffer) >= self.buffer_size
 
-    @task(name="Flush")
     def flush(self):
         """
             Sends all events in the buffer to the collector.
@@ -387,6 +385,14 @@ def consume(self):
             self.queue.task_done()
 
 
+@app.task(bind=True, name='tasks.flush')  # the self passed with bind can be used for on_fail/retrying
+def flush_emitter(self, emitter):
+    try:
+        emitter.flush()
+    finally:
+        logger.info("Flush called on emitter")
+
+
 class CeleryEmitter(Emitter):
     """
         Uses a Celery worker to send HTTP requests asynchronously.
@@ -400,7 +406,7 @@ def flush(self):
         """
             Schedules a flush task
         """
-        super(CeleryEmitter, self).flush.delay()
+        flush_emitter.delay(self)  # passes emitter (self - CeleryEmitter) to task
         logger.info("Scheduled a Celery task to flush the event queue")