add play, sniff and --filter to consume&listen
#37
Conversation
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
| ## `sniff` - List all topics and their high, low watermarks and number of messages | ||
| `saluki sniff mybroker:9092` |
There was a problem hiding this comment.
I intuitively typed saluki sniff livedata.isis.cclrc.ac.uk:31092/NDW2922_sampleEnv which sort of worked but actually listed all topics. Given how many topics exist, it would be nice for that to work and just give me the watermarks of that one topic?
There was a problem hiding this comment.
nah just use grep m8
There was a problem hiding this comment.
but yeh alright i'll do it
|
|
||
| ### Between timestamps | ||
|
|
||
| `saluki play mybroker:9092/source_topic mybroker:9092/dest_topic -t 1762209990 1762209992` - This will forward messages between the two given timestamps. |
There was a problem hiding this comment.
So I think this should have worked then?
(.venv) c:\Instrument\dev\saluki>saluki play livedata.isis.cclrc.ac.uk:31092/NDW2922_bluesky livedata.isis.cclrc.ac.uk:31092/TOMTEST_bluesky -t 1762972997 1762979997
Traceback (most recent call last):
File "<frozen runpy>", line 198, in _run_module_as_main
File "<frozen runpy>", line 88, in _run_code
File "c:\Instrument\dev\saluki\.venv\Scripts\saluki.exe\__main__.py", line 7, in <module>
File "C:\Instrument\dev\saluki\src\saluki\main.py", line 124, in main
play(
File "C:\Instrument\dev\saluki\src\saluki\play.py", line 45, in play
TopicPartition(src_topic, src_partition, timestamps[0]),
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: 'str' object cannot be interpreted as an integer
There was a problem hiding this comment.
Everywhere where we support timestamps, do we also want to support passing in a human-format time and parsing it to a timestamp?
I'd kind of like to be able to saluki consume some_topic -t "2025-11-11 09:00" -m 100 to consume 100 messages after 9 o'clock on 11/11... without typing 1762851600
| `saluki consume mybroker:9092/mytopic -p 1 -o 123456 -m 10` - This will print 9 messages before (and inclusively the offset specified) offset `123456` of `mytopic` on `mybroker`, in partition 1. | ||
|
|
||
| Use the `-g` flag to go the other way, ie. in the above example to consume the 9 messages _after_ offset 123456 | ||
|
|
||
| # Install | ||
| `pip install saluki` | ||
| You can also filter out messages to specific schema(s) with the `-f` flag, like the example above for `listen`. |
There was a problem hiding this comment.
Maybe this should also support timestamps - offsets are not very user friendly...
rerpha
moved this to In progress
in Data streaming for HRPD-X (and possibly SANDALS-II)
rerpha
moved this from In progress
to In review
in Data streaming for HRPD-X (and possibly SANDALS-II)
Tom-Willemsen
left a comment
Tom-Willemsen
left a comment
There was a problem hiding this comment.
Thanks - this looks to be working nicely now.
github-project-automation
bot
moved this from In review
to Done
in Data streaming for HRPD-X (and possibly SANDALS-II)
closes #33
This adds
sniff,play, and filtering toconsumeandlistenas described in theREADME.md. When this is merged I'll create a new release.