Renamed consumers to emitters (snowplow#90)

Author: fblundun

snowplow_tracker/__init__.py

@@ -1,5 +1,5 @@
 from snowplow_tracker._version import __version__
 from snowplow_tracker.subject import Subject
-from snowplow_tracker.consumer import logger, Consumer, AsyncConsumer, CeleryConsumer, RedisConsumer
+from snowplow_tracker.emitters import logger, Emitter, AsyncEmitter, CeleryEmitter, RedisEmitter
 from snowplow_tracker.tracker import Tracker
 from contracts import disable_all as disable_contracts, enable_all as enable_contracts

snowplow_tracker/consumer.py snowplow_tracker/emitters.pysnowplow_tracker/consumer.py renamed to snowplow_tracker/emitters.py

@@ -1,5 +1,5 @@
 """
-    consumer.py
+    emitters.py
 
     Copyright (c) 2013-2014 Snowplow Analytics Ltd. All rights reserved.
 
@@ -55,7 +55,7 @@
     app = Celery("Snowplow", broker="redis://guest@localhost//")
 
 
-class Consumer(object):
+class Emitter(object):
     """
         Synchronously send Snowplow events to a Snowplow as_collector_uri
         Supports both GET and POST requests
@@ -84,7 +84,7 @@ def __init__(self, endpoint, protocol="http", port=None, method="get", buffer_si
                                    If method is "get":  An array of dictionaries corresponding to the unsent events' payloads
             :type  on_failure:  function | None            
         """
-        self.endpoint = Consumer.as_collector_uri(endpoint, protocol, port)
+        self.endpoint = Emitter.as_collector_uri(endpoint, protocol, port)
 
         self.method = method
 
@@ -200,45 +200,45 @@ def http_get(self, payload):
 
     def sync_flush(self):
         """
-            Calls the flush method of the base Consumer class.
+            Calls the flush method of the base Emitter class.
             This is guaranteed to be blocking, not asynchronous.
         """
         logger.debug("Starting synchronous flush...")
-        result = Consumer.flush(self)
+        result = Emitter.flush(self)
         for t in self.threads:
             t.join(THREAD_TIMEOUT)
         logger.debug("Finished synchrous flush")
         return result
 
 
-class AsyncConsumer(Consumer):
+class AsyncEmitter(Emitter):
     """
         Uses threads to send HTTP requests asynchronously
     """
     def flush(self):
         """
             Removes all dead threads, then creates a new thread which
-            excecutes the flush method of the base Consumer class
+            excecutes the flush method of the base Emitter class
         """
         self.threads = [t for t in self.threads if t.isAlive()]
         logger.debug("Flushing thread running...")
-        t = threading.Thread(target=super(AsyncConsumer, self).flush)
+        t = threading.Thread(target=super(AsyncEmitter, self).flush)
         self.threads.append(t)
         t.start()
 
 
-class CeleryConsumer(Consumer):
+class CeleryEmitter(Emitter):
     """
         Uses a Celery worker to send HTTP requests asynchronously
     """
     def flush(self):
         """
             Schedules a flush task
         """
-        super(CeleryConsumer, self).flush.delay()
+        super(CeleryEmitter, self).flush.delay()
 
 
-class RedisConsumer(object):
+class RedisEmitter(object):
     """
         Sends Snowplow events to a Redis database
     """
@@ -266,7 +266,7 @@ def input(self, payload):
         logger.debug("Finished sending event to Redis.")
 
     def flush(self):
-        logger.warn("The RedisConsumer class does not need to be flushed")
+        logger.warn("The RedisEmitter class does not need to be flushed")
 
     def sync_flush(self):
         self.flush()

snowplow_tracker/test/integration/test_integration.py

@@ -25,7 +25,7 @@
 import redis
 import json
 import base64
-from snowplow_tracker import tracker, _version, consumer, subject
+from snowplow_tracker import tracker, _version, emitters, subject
 from httmock import all_requests, HTTMock
 
 try:
@@ -36,7 +36,7 @@
 
 querystrings = [""]
 
-default_consumer = consumer.Consumer("localhost", protocol="http", port=80)
+default_emitter = emitters.Emitter("localhost", protocol="http", port=80)
 
 default_subject = subject.Subject()
 
@@ -65,23 +65,23 @@ def fail_response_content(url, request):
 class IntegrationTest(unittest.TestCase):
 
     def test_integration_page_view(self):
-        t = tracker.Tracker(default_consumer, default_subject)
+        t = tracker.Tracker(default_emitter, default_subject)
         with HTTMock(pass_response_content):
             t.track_page_view("http://savethearctic.org", "Save The Arctic", "http://referrer.com")
         expected_fields = {"e": "pv", "page": "Save+The+Arctic", "url": "http%3A%2F%2Fsavethearctic.org", "refr": "http%3A%2F%2Freferrer.com"}
         for key in expected_fields:
             self.assertEquals(from_querystring(key, querystrings[-1]), expected_fields[key])            
 
     def test_integration_ecommerce_transaction_item(self):
-        t = tracker.Tracker(default_consumer, default_subject)
+        t = tracker.Tracker(default_emitter, default_subject)
         with HTTMock(pass_response_content):
             t.track_ecommerce_transaction_item("12345", "pbz0025", 7.99, 2, "black-tarot", "tarot", currency="GBP")
         expected_fields = {"ti_ca": "tarot", "ti_id": "12345", "ti_qu": "2", "ti_sk": "pbz0025", "e": "ti", "ti_nm": "black-tarot", "ti_pr": "7.99", "ti_cu": "GBP"}
         for key in expected_fields:
             self.assertEquals(from_querystring(key, querystrings[-1]), expected_fields[key])
 
     def test_integration_ecommerce_transaction(self):
-        t = tracker.Tracker(default_consumer, default_subject)
+        t = tracker.Tracker(default_emitter, default_subject)
         with HTTMock(pass_response_content):
             t.track_ecommerce_transaction("6a8078be", 35, city="London", currency="GBP", items=
                 [{  
@@ -110,23 +110,23 @@ def test_integration_ecommerce_transaction(self):
         self.assertEquals(from_querystring("dtm", querystrings[-3]), from_querystring("dtm", querystrings[-2]))
 
     def test_integration_screen_view(self):
-        t = tracker.Tracker(default_consumer, default_subject)
+        t = tracker.Tracker(default_emitter, default_subject)
         with HTTMock(pass_response_content):
             t.track_screen_view("Game HUD 2", "Hello!")
         expected_fields = {"e": "ue"}
         for key in expected_fields:          
             self.assertEquals(from_querystring(key, querystrings[-1]), expected_fields[key])
 
     def test_integration_struct_event(self):
-        t = tracker.Tracker(default_consumer, default_subject)
+        t = tracker.Tracker(default_emitter, default_subject)
         with HTTMock(pass_response_content):
             t.track_struct_event("Ecomm", "add-to-basket", "dog-skateboarding-video", "hd", 13.99)
         expected_fields = {"se_ca": "Ecomm", "se_pr": "hd", "se_la": "dog-skateboarding-video", "se_va": "13.99", "se_ac": "add-to-basket", "e": "se"}
         for key in expected_fields:
             self.assertEquals(from_querystring(key, querystrings[-1]), expected_fields[key])
 
     def test_integration_unstruct_event_non_base64(self):
-        t = tracker.Tracker(default_consumer, default_subject, encode_base64=False)
+        t = tracker.Tracker(default_emitter, default_subject, encode_base64=False)
         with HTTMock(pass_response_content):
             t.track_unstruct_event({"schema": "com.acme/viewed_product/json/2-0-2", "data": {"product_id": "ASO01043", "price$flt": 49.95, "walrus$tms": 1000}})
         expected_fields = {"e": "ue"}
@@ -140,21 +140,21 @@ def test_integration_unstruct_event_non_base64(self):
         })
 
     def test_integration_unstruct_event_base64(self):
-        t = tracker.Tracker(default_consumer, default_subject, encode_base64=True)
+        t = tracker.Tracker(default_emitter, default_subject, encode_base64=True)
         with HTTMock(pass_response_content):
             t.track_unstruct_event({"schema": "com.acme/viewed_product/json/2-0-2", "data": {"product_id": "ASO01043", "price$flt": 49.95, "walrus$tms": 1000}})
         expected_fields = {"e": "ue"}
         for key in expected_fields:
             self.assertEquals(from_querystring(key, querystrings[-1]), expected_fields[key])
         envelope_string = from_querystring("ue_px", querystrings[-1])
-        envelope = json.loads((base64.urlsafe_b64decode(bytearray(envelope_string, 'utf-8'))).decode("utf-8"))
+        envelope = json.loads((base64.urlsafe_b64decode(bytearray(envelope_string, "utf-8"))).decode("utf-8"))
         self.assertEquals(envelope, {
             "schema": "com.snowplowanalytics/unstruct_event/json/1-0-0",
             "data": {"schema": "com.acme/viewed_product/json/2-0-2", "data": {"product_id": "ASO01043", "price$flt": 49.95, "walrus$tms": 1000}}
         })
 
     def test_integration_context_non_base64(self):
-        t = tracker.Tracker(default_consumer, default_subject, encode_base64=False)
+        t = tracker.Tracker(default_emitter, default_subject, encode_base64=False)
         with HTTMock(pass_response_content):
             t.track_page_view("localhost", "local host", None, [{"schema": "com.example/user/json/2-0-3", "data": {"user_type": "tester"}}])
         envelope_string = from_querystring("co", querystrings[-1])
@@ -165,11 +165,11 @@ def test_integration_context_non_base64(self):
         })
 
     def test_integration_context_base64(self):
-        t = tracker.Tracker(default_consumer, default_subject, encode_base64=True)
+        t = tracker.Tracker(default_emitter, default_subject, encode_base64=True)
         with HTTMock(pass_response_content):
             t.track_page_view("localhost", "local host", None, [{"schema": "com.example/user/json/2-0-3", "data": {"user_type": "tester"}}])
         envelope_string = from_querystring("cx", querystrings[-1])
-        envelope = json.loads((base64.urlsafe_b64decode(bytearray(envelope_string, 'utf-8'))).decode("utf-8"))
+        envelope = json.loads((base64.urlsafe_b64decode(bytearray(envelope_string, "utf-8"))).decode("utf-8"))
         self.assertEquals(envelope, {
             "schema": "com.snowplowanalytics/contexts/json/1-0-0",
             "data":[{"schema": "com.example/user/json/2-0-3", "data": {"user_type": "tester"}}]
@@ -184,7 +184,7 @@ def test_integration_standard_nv_pairs(self):
         s.set_timezone("Europe London")
         s.set_lang("en")
 
-        t = tracker.Tracker(consumer.Consumer("localhost"), s, "cf", app_id="angry-birds-android")
+        t = tracker.Tracker(emitters.Emitter("localhost"), s, "cf", app_id="angry-birds-android")
         with HTTMock(pass_response_content):
             t.track_page_view("localhost", "local host")
         expected_fields = {"tna": "cf", "res": "100x200",
@@ -195,15 +195,15 @@ def test_integration_standard_nv_pairs(self):
 
     def test_integration_redis_default(self):
         r = redis.StrictRedis()
-        t = tracker.Tracker(consumer.RedisConsumer(), default_subject)
+        t = tracker.Tracker(emitters.RedisEmitter(), default_subject)
         t.track_page_view("http://www.example.com")
         event_string = r.rpop("snowplow")
         event_dict = json.loads(event_string.decode("utf-8"))
         self.assertEquals(event_dict["e"], "pv")
 
     def test_integration_redis_custom(self):
         r = redis.StrictRedis(db=1)
-        t = tracker.Tracker(consumer.RedisConsumer(rdb=r, key="custom_key"), default_subject)
+        t = tracker.Tracker(emitters.RedisEmitter(rdb=r, key="custom_key"), default_subject)
         t.track_page_view("http://www.example.com")
         event_string = r.rpop("custom_key")
         event_dict = json.loads(event_string.decode("utf-8"))
@@ -212,9 +212,9 @@ def test_integration_redis_custom(self):
     def test_integration_success_callback(self):
         callback_success_queue = []
         callback_failure_queue = []
-        callback_consumer = consumer.Consumer("localhost", on_success=lambda x: callback_success_queue.append(x),
+        callback_emitter = emitters.Emitter("localhost", on_success=lambda x: callback_success_queue.append(x),
                                                            on_failure=lambda x, y:callback_failure_queue.append(x))
-        t = tracker.Tracker(callback_consumer, default_subject)
+        t = tracker.Tracker(callback_emitter, default_subject)
         with HTTMock(pass_response_content):
             t.track_page_view("http://www.example.com")
         self.assertEquals(callback_success_queue[0], 1)
@@ -223,9 +223,9 @@ def test_integration_success_callback(self):
     def test_integration_failure_callback(self):
         callback_success_queue = []
         callback_failure_queue = []
-        callback_consumer = consumer.Consumer("localhost", on_success=lambda x: callback_success_queue.append(x),
+        callback_emitter = emitters.Emitter("localhost", on_success=lambda x: callback_success_queue.append(x),
                                                            on_failure=lambda x, y:callback_failure_queue.append(x))
-        t = tracker.Tracker(callback_consumer, default_subject)
+        t = tracker.Tracker(callback_emitter, default_subject)
         with HTTMock(fail_response_content):
             t.track_page_view("http://www.example.com")
         self.assertEquals(callback_success_queue, [])

snowplow_tracker/test/unit/test_tracker.py

@@ -23,7 +23,7 @@
 import unittest
 from freezegun import freeze_time
 from snowplow_tracker.tracker import Tracker
-from snowplow_tracker.consumer import Consumer
+from snowplow_tracker.emitters import Emitter
 
 
 class TestTracker(unittest.TestCase):
@@ -32,7 +32,7 @@ def setUp(self):
         pass
 
     def test_initialisation(self):
-        t = Tracker(Consumer("d3rkrsqld9gmqf.cloudfront.net"), namespace="cloudfront", encode_base64= False, app_id="AF003")
+        t = Tracker(Emitter("d3rkrsqld9gmqf.cloudfront.net"), namespace="cloudfront", encode_base64= False, app_id="AF003")
         self.assertEquals(t.standard_nv_pairs["tna"], "cloudfront")
         self.assertEquals(t.standard_nv_pairs["aid"], "AF003")
         self.assertEquals(t.encode_base64, False)

snowplow_tracker/tracker.py

@@ -51,14 +51,14 @@ class Tracker:
 
     new_contract("tracker", lambda s: isinstance(s, Tracker))
 
-    new_contract("consumer", lambda s: hasattr(s, "input"))
+    new_contract("emitter", lambda s: hasattr(s, "input"))
 
     @contract
     def __init__(self, out_queue, _subject=None,
                  namespace=None, app_id=None, encode_base64=DEFAULT_ENCODE_BASE64):
         """
-            :param out_queue:        Consumer to which events will be sent
-            :type  out_queue:        consumer
+            :param out_queue:        Emitter to which events will be sent
+            :type  out_queue:        emitter
             :param _subject:         Subject to be tracked
             :type  _subject:         subject | None
             :param namespace:        Identifier for the Tracker instance
@@ -115,7 +115,7 @@ def get_timestamp(tstamp=None):
     @contract
     def track(self, pb):
         """
-            Send the payload to a consumer
+            Send the payload to a emitter
 
             :param  pb:              Payload builder
             :type   pb:              payload
@@ -350,7 +350,7 @@ def track_unstruct_event(self, event_json, context=None, tstamp=None):
     @contract
     def flush(self, async=False):
         """
-            Flush the consumer
+            Flush the emitter
 
             :param  async:  Whether the flush is done asynchronously. Default is False
             :type   async:  bool