Allowed a single Tracker instance to send events to multiple Emitters

- closes snowplow#91

Author: fblundun

CHANGES.txt

@@ -3,10 +3,9 @@ Version 0.5.0 (2014-08-xx)
 Converted payload values to strings for POST requests (#100)
 Set content type to "application/json; charset=utf-8" for POST requests (#99)
 Changed collector endpoint for POST to /com.snowplowanalytics.snowplow/tp2 (#98)
-Added ability to batch based on payload size (#95) (TODO)
 Stopped setting and sending tid (#94)
 Started setting and sending eid (#93)
-Allowed a single Tracker instance to send events to multiple Emitters (#91) (TODO)
+Allowed a single Tracker instance to send events to multiple Emitters (#91)
 
 Version 0.4.0 (2014-06-10)
 --------------------------

snowplow_tracker/test/integration/test_integration.py

@@ -65,23 +65,23 @@ def fail_response_content(url, request):
 class IntegrationTest(unittest.TestCase):
 
     def test_integration_page_view(self):
-        t = tracker.Tracker(default_emitter, 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_emitter, 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_emitter, 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_emitter, 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_emitter, 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_emitter, 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": "iglu:com.acme/viewed_product/jsonschema/2-0-2", "data": {"product_id": "ASO01043", "price$flt": 49.95, "walrus$tms": 1000}})
         expected_fields = {"e": "ue"}
@@ -140,7 +140,7 @@ def test_integration_unstruct_event_non_base64(self):
         })
 
     def test_integration_unstruct_event_base64(self):
-        t = tracker.Tracker(default_emitter, 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": "iglu:com.acme/viewed_product/jsonschema/2-0-2", "data": {"product_id": "ASO01043", "price$flt": 49.95, "walrus$tms": 1000}})
         expected_fields = {"e": "ue"}
@@ -154,7 +154,7 @@ def test_integration_unstruct_event_base64(self):
         })
 
     def test_integration_context_non_base64(self):
-        t = tracker.Tracker(default_emitter, 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": "iglu:com.example/user/jsonschema/2-0-3", "data": {"user_type": "tester"}}])
         envelope_string = from_querystring("co", querystrings[-1])
@@ -165,7 +165,7 @@ def test_integration_context_non_base64(self):
         })
 
     def test_integration_context_base64(self):
-        t = tracker.Tracker(default_emitter, 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": "iglu:com.example/user/jsonschema/2-0-3", "data": {"user_type": "tester"}}])
         envelope_string = unquote_plus(from_querystring("cx", querystrings[-1]))
@@ -184,7 +184,7 @@ def test_integration_standard_nv_pairs(self):
         s.set_timezone("Europe London")
         s.set_lang("en")
 
-        t = tracker.Tracker(emitters.Emitter("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",
@@ -197,15 +197,15 @@ def test_integration_standard_nv_pairs(self):
 
     def test_integration_redis_default(self):
         r = redis.StrictRedis()
-        t = tracker.Tracker(emitters.RedisEmitter(), 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(emitters.RedisEmitter(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"))
@@ -216,7 +216,7 @@ def test_integration_success_callback(self):
         callback_failure_queue = []
         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_emitter, 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)
@@ -227,7 +227,7 @@ def test_integration_failure_callback(self):
         callback_failure_queue = []
         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_emitter, 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

@@ -33,7 +33,7 @@ def setUp(self):
         pass
 
     def test_initialisation(self):
-        t = Tracker(Emitter("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

@@ -55,11 +55,11 @@ class Tracker:
     new_contract("emitter", lambda s: hasattr(s, "input"))
 
     @contract
-    def __init__(self, emitter, subject=None,
+    def __init__(self, emitters, subject=None,
                  namespace=None, app_id=None, encode_base64=DEFAULT_ENCODE_BASE64):
         """
-            :param emitter:          Emitter to which events will be sent
-            :type  emitter:          emitter
+            :param emitters:         Emitters to which events will be sent
+            :type  emitters:         list[>0](emitter)
             :param subject:          Subject to be tracked
             :type  subject:          subject | None
             :param namespace:        Identifier for the Tracker instance
@@ -72,7 +72,7 @@ def __init__(self, emitter, subject=None,
         if subject is None:
             subject = _subject.Subject()
 
-        self.emitter = emitter
+        self.emitters = emitters
         self.subject = subject
         self.encode_base64 = encode_base64
 
@@ -119,11 +119,12 @@ def track(self, pb):
 
             :param  pb:              Payload builder
             :type   pb:              payload
-            :rtype:                  tracker | int
+            :rtype:                  tracker | list(int)
         """
-        result = self.emitter.input(pb.nv_pairs)
-        if result is not None:
-            return result
+        results = [emitter.input(pb.nv_pairs) for emitter in self.emitters]
+        filtered = [res for res in results if res is not None]
+        if filtered:
+            return filtered
         else:
             return self
 
@@ -139,7 +140,7 @@ def complete_payload(self, pb, context, tstamp):
             :type   context:         list(dict(string:*)) | None
             :param  tstamp:          Optional user-provided timestamp for the event
             :type   tstamp:          int | float | None
-            :rtype:                  tracker | int
+            :rtype:                  tracker | list(int)
         """
         pb.add("eid", Tracker.get_uuid())
         pb.add("dtm", Tracker.get_timestamp(tstamp))
@@ -164,7 +165,7 @@ def track_page_view(self, page_url, page_title=None, referrer=None, context=None
             :type   referrer:       string_or_none
             :param  context:        Custom context for the event
             :type   context:        list(dict(string:*)) | None
-            :rtype:                 tracker | int
+            :rtype:                 tracker | list(int)
         """
         pb = payload.Payload()
         pb.add("e", "pv")           # pv: page view
@@ -199,7 +200,7 @@ def track_ecommerce_transaction_item(self, order_id, sku, price, quantity,
             :type   currency:    string_or_none
             :param  context:     Custom context for the event
             :type   context:     list(dict(string:*)) | None
-            :rtype:              tracker | int
+            :rtype:              tracker | list(int)
         """
         pb = payload.Payload()
         pb.add("e", "ti")
@@ -282,7 +283,7 @@ def track_screen_view(self, name, id_=None, context=None, tstamp=None):
             :type   id_:            string_or_none
             :param  context:        Custom context for the event
             :type   context:        list(dict(string:*)) | None
-            :rtype:                 tracker | int
+            :rtype:                 tracker | list(int)
         """
         screen_view_properties = {"name": name}
         if id_ is not None:
@@ -313,7 +314,7 @@ def track_struct_event(self, category, action, label=None, property_=None, value
             :type   value:          int | float | None
             :param  context:        Custom context for the event
             :type   context:        list(dict(string:*)) | None
-            :rtype:                 tracker | int
+            :rtype:                 tracker | list(int)
         """
         pb = payload.Payload()
         pb.add("e", "se")
@@ -335,7 +336,7 @@ def track_unstruct_event(self, event_json, context=None, tstamp=None):
             :type   event_json:      dict(string: string | dict)
             :param  context:         Custom context for the event
             :type   context:         list(dict(string:*)) | None
-            :rtype:                  tracker | int
+            :rtype:                  tracker | list(int)
         """
 
         envelope = {"schema": UNSTRUCT_EVENT_SCHEMA , "data": event_json}
@@ -354,7 +355,7 @@ def flush(self, async=False):
 
             :param  async:  Whether the flush is done asynchronously. Default is False
             :type   async:  bool
-            :rtype:         tracker | int
+            :rtype:         tracker | list(int)
         """
         if async:
             self.emitter.flush()