Consumer topics sampling

[diorcety]

Hello,
I have a special need, I want to sample a stream according the consumer . For example every X seconds only keep the last msg for each topic. It doesn't seem possible currently, no? In which part of code I have to look to implement such behaviour?
Thanks

[derekcollison]

Not sure we have enough information about what you are looking at doing.

Could we get more information about the stream (NATS CLI nats str info <STREAM> and the consumer nats consumer info <STREAM> <CONSUMER>?

Streams have the ability to limit the number of msgs per subject, where MaxMsgsPerSubject of 1 only keeps the last value. This is how we do Key-Value materialized views.

Streams can also republish msgs after they are timestamped and sequenced. You can republish on any subject transform you want. You could then create a stream or a KV with the subject filter that you want to capture and that will keep the last value.

However you can always ask a stream for the last msg per subject which does not require a consumer.

[diorcety]

I have a "realtime" stream, containing high frequency publications on multiple topics (only the last one is saved). I would want create a new stream with this one as source, reducing the frequency by emitting only one message during a window of X seconds, and save it for consumer. Like if you are using the sample function on reactivex : (https://reactivex.io/documentation/operators/sample.html).
This stream will be used off site so I would minimize the bandwidth (and storage on site) and avoid doing this filter on the client side.
Currently there is a way to filter by subject, I would try to filter by sampling.

[derekcollison]

Create a stream that captures the filtered subject and has MaxMsgPerSubject of 1. Within whatever smapling frequency you want, create a consumer and whip through all the latest values and republish as needed. Or let applications sample the latest values from the stream via GetLastBySubject calls at the sample frequency.

[diorcety]

Replace the source of the second stream by an application, reading on first stream, time filtering and pushing on second stream? That sound like a good feature to include some kind of linq in order to allow , subject filtering (currently doable), time filtering and other features, without to develop an app that. Some plumping features.

[derekcollison]

Just need one stream that has MaxMsgsPer for how many historical values per subject you need. App can ask for LastPerSubject at whatever interval it desires.

[diorcety]

No, because I use the second stream with retention policy. In case of connection issue I need to get all the history (in this case with time filtering). The first stream is for high frequency non persisted local clients and the second one for remote clients which need all the history at lower frequencies.