add skeleton for play()

Author: rerpha

src/saluki/consume.py

@@ -29,7 +29,7 @@ def consume(
     c = Consumer(
         {
             "bootstrap.servers": broker,
-            "group.id": "saluki",
+            "group.id": "saluki-consume",
             "session.timeout.ms": 6000,
             "auto.offset.reset": "latest",
             "enable.auto.offset.store": False,

src/saluki/main.py

@@ -4,6 +4,7 @@
 
 from saluki.consume import consume
 from saluki.listen import listen
+from saluki.play import play
 from saluki.sniff import sniff
 from saluki.utils import parse_kafka_uri
 
@@ -88,7 +89,8 @@ def main() -> None:
     # saluki consume -f pl72,6s4t mybroker:9092/XXX_runInfo -m 10 # get the last pl72 run starts or 6s4t run stops#
 
     # saluki bury - dump data on topic to file
-    # saluki bury mybroker:9092/topicname -p 0 -f offsetortimestamp -t offsetortimestamp outputfile
+    # saluki bury mybroker:9092/topicname -p 0 -o startoffset finishoffset outputfile
+    # saluki bury mybroker:9092/topicname -p 0 -t starttimestamp finishtimestamp outputfile
 
     # saluki dig - push data from dump generated by saluki bury to topic 
     # saluki dig mybroker:9092/topicname -p 0 outputfile
@@ -102,6 +104,7 @@ def main() -> None:
     g = play_parser.add_mutually_exclusive_group(required=True)
     g.add_argument("-o", "--offsets", help="offsets to replay between (inclusive)", type=int, nargs=2)
     g.add_argument("-t", "--timestamps", help="timestamps to replay between", type=str, nargs=2)
+    g.add_argument("-c", "--chunk", help="forward in chunks. ie to avoid storing a huge list in memory", default=0, type=int, required=False)
 
     if len(sys.argv) == 1:
         parser.print_help()
@@ -135,7 +138,7 @@ def main() -> None:
             print(f"Replaying {src_broker}/{src_topic} between timestamps {args.timestamps[0]} and {args.timestamps[1]} to {dest_broker}/{dest_topic}")
 
         if input("OK? (y/n)").lower() == 'y':
-            #play(src_broker, src_topic, dest_broker, dest_topic, args.offset, args.timestamp)
+            play(src_broker, src_topic, dest_broker, dest_topic, args.offsets, args.timestamps, args.chunks)
             print("replayed")
     elif args.command == _SNIFF:
         sniff(args.broker)

src/saluki/play.py

@@ -0,0 +1,12 @@
+from confluent_kafka import Consumer, Producer
+
+
+def play(src_broker: str, src_topic: str, dest_broker: str, dest_topic: str, offsets:list[int]|None, timestamps:list[int]|None, chunks:int) -> None:
+    consumer = Consumer(        {
+            "bootstrap.servers": src_broker,
+            "group.id": "saluki-play",
+        })
+    producer = Producer({
+        "bootstrap.servers": dest_broker,
+    })
+    pass