added rolling data producer

Author: ftisiot

Diff for: README.md

@@ -33,7 +33,8 @@ python main.py \
   --port 13041 \
   --topic-name pizza-orders \
   --nr-messages 0 \
-  --max-waiting-time 0
+  --max-waiting-time 0 \
+  --subject pizza
 ```
 in ``SASL_SSL`` security protocol:
 ```bash
@@ -47,7 +48,8 @@ python main.py \
   --port 13041 \
   --topic-name pizza-orders \
   --nr-messages 0 \
-  --max-waiting-time 0
+  --max-waiting-time 0 \
+  --subject pizza
 ```
 in ``PLAINTEXT`` security protocol:
 ```bash
@@ -57,7 +59,8 @@ python main.py \
   --port 9092 \
   --topic-name pizza-orders \
   --nr-messages 0 \
-  --max-waiting-time 0
+  --max-waiting-time 0 \
+  --subject pizza
 ```
 Where
 * `security-protocol`: Security protocol for Kafka. ``PLAINTEXT``,  ``SSL`` or ``SASL_SSL`` are supported.
@@ -67,7 +70,7 @@ Where
 * `topic-name`: the Apache Kafka topic name to write to (the topic needs to be pre-created or `kafka.auto_create_topics_enable` parameter enabled)
 * `nr-messages`: the number of messages to send
 * `max-waiting-time`: the maximum waiting time in seconds between messages
-* `subject`: select amongst various subjects: `pizza` is the default one, but you can generate also `userbehaviour`, `bet`, `stock`, `realstock` (using the yahoo finance apis) and `metric`.
+* `subject`: select amongst various subjects: `pizza` is the default one, but you can generate also `userbehaviour`, `bet`, `stock`, `realstock` (using the yahoo finance apis), `metric`, and `rolling`.
 
 If successfully connected to a Apache Kafka cluster, the command will output a number of messages (`nr-messages` parameter) that are been sent to Apache Kafka in the form
 

Diff for: data_rolling.json

@@ -11,6 +11,7 @@
 from realstockproducer import RealStockProvider
 from metricproducer import MetricProvider
 from userbets import UserBetsProvider
+from rolling import RollingProvider
 
 
 MAX_NUMBER_PIZZAS_IN_ORDER = 10
@@ -79,10 +80,12 @@ def produce_msgs(security_protocol='SSL',
         fake.add_provider(UserBehaviorProvider)
     elif subject == 'bet':
         fake.add_provider(UserBetsProvider)
+    elif subject == 'rolling':
+        fake.add_provider(RollingProvider)
     else:
         fake.add_provider(PizzaProvider)
     while i < nr_messages:
-        if subject in ['stock', 'userbehaviour', 'realstock', 'metric','bet']:
+        if subject in ['stock', 'userbehaviour', 'realstock', 'metric', 'bet', 'rolling']:
             message, key = fake.produce_msg()
         else:
             message, key = fake.produce_msg(fake, i, MAX_NUMBER_PIZZAS_IN_ORDER, MAX_ADDITIONAL_TOPPINGS_IN_PIZZA)

Diff for: main.py

@@ -0,0 +1,36 @@
+import random
+import time
+from faker.providers import BaseProvider
+import json
+
+
+nr_item = 0
+base_timestamp = int(time.time()*1000)
+f = open('data_rolling.json')
+data = json.load(f)
+
+# Adding a RollingProvider with 1 method:
+#   * getcode to retrieve the username
+
+class RollingProvider(BaseProvider):
+
+    def produce_msg (self):
+        
+        global nr_item
+        global base_timestamp
+        item = data[nr_item]        
+
+        # message composition
+        message = {
+            'val': item['value'],
+            'ts': base_timestamp+int(item['timestamp'])
+        }
+        key = {'ts':base_timestamp+int(item['timestamp'])}
+        if nr_item == 1732:
+            print("Waiting next iteration")
+            time.sleep(30)
+            nr_item = 0
+            base_timestamp = int(time.time()*1000)
+        else:
+            nr_item = nr_item + 1
+        return message, key