Subscribing to data

from quixstreams import KafkaStreamingClient

client = KafkaStreamingClient('127.0.0.1:9092')

var client = new QuixStreams.Streaming.KafkaStreamingClient("127.0.0.1:9092");

topic_consumer = client.get_topic_consumer(TOPIC)

var topicConsumer = client.GetTopicConsumer(TOPIC);

topic_consumer = client.get_topic_consumer("{topic}","{your-consumer-group-id}")

var topicConsumer = client.GetTopicConsumer("{topic}","{your-consumer-group-id}");

from quixstreams import TopicConsumer, StreamConsumer

def on_stream_received_handler(stream_received: StreamConsumer):
    print("Stream received:" + stream_received.stream_id)

topic_consumer.on_stream_received = on_stream_received_handler
topic_consumer.subscribe()

topicConsumer.OnStreamReceived += (topic, newStream) =>
{
    Console.WriteLine($"New stream read: {newStream.StreamId}");
};

topicConsumer.Subscribe();

from quixstreams import TopicConsumer, StreamConsumer, TimeseriesData

def on_stream_received_handler(stream_received: StreamConsumer):
    stream_received.timeseries.on_data_received = on_timeseries_data_received_handler

def on_timeseries_data_received_handler(stream: StreamConsumer, data: TimeseriesData):
    with data:
        timestamp = data.timestamps[0].timestamp
        num_value = data.timestamps[0].parameters['ParameterA'].numeric_value
        print("ParameterA - " + str(timestamp) + ": " + str(num_value))

topic_consumer.on_stream_received = on_stream_received_handler
topic_consumer.subscribe()

topicConsumer.OnStreamReceived += (topic, streamConsumer) =>
{
    streamConsumer.Timeseries.OnDataReceived += (sender, args) =>
    {
        var timestamp = args.Data.Timestamps[0].Timestamp;
        var numValue = args.Data.Timestamps[0].Parameters["ParameterA"].NumericValue;
        Console.WriteLine($"ParameterA - {timestamp}: {numValue}");
    };
};

topicConsumer.Subscribe();

for ts in data.timestamps:
    timestamp = ts.timestamp_nanoseconds
    numValue = ts.parameters['Speed'].numeric_value
    print("Speed - " + str(timestamp) ": " + str(numValue))

foreach (var timestamp in data.Timestamps)
{
       var timestamp = timestamp.TimestampNanoseconds;
       var numValue = timestamp.Parameters["Speed"].NumericValue;
       Console.WriteLine($"Speed - {timestamp}: {numValue}");
}

buffer = newStream.timeseries.create_buffer()

var buffer = newStream.Timeseries.CreateBuffer();

buffer.time_span_in_milliseconds = 100

buffer.TimeSpanInMilliseconds = 100;

from quixstreams import TopicConsumer, StreamConsumer, TimeseriesData

def on_data_released_handler(stream: StreamConsumer, data: TimeseriesData):
    with data:
        timestamp = data.timestamps[0].timestamp
        num_value = data.timestamps[0].parameters['ParameterA'].numeric_value
        print("ParameterA - " + str(timestamp) + ": " + str(num_value))

buffer.on_data_released = on_data_released_handler
# buffer.on_dataframe_released and other callbacks are also available, check reading without buffer for more info

buffer.OnDataReleased += (sender, args) =>
{
    var timestamp = ags.Data.Timestamps[0].Timestamp;
    var numValue = ags.Data.Timestamps[0].Parameters["ParameterA"].NumericValue;
    Console.WriteLine($"ParameterA - {timestamp}: {numValue}");
};

stream.timeseries.buffer.packet_size = 100
stream.timeseries.buffer.buffer_timeout = 1000

stream.Timeseries.Buffer.PacketSize = 100;
stream.Timeseries.Buffer.BufferTimeout = 1000;

buffer.time_span_in_milliseconds = 100
buffer.custom_trigger = lambda data: data.timestamps[0].tags["is_critical"] == 'True'

stream.Timeseries.Buffer.TimeSpanInMilliseconds = 100;
stream.Timeseries.Buffer.CustomTrigger = data => data.Timestamps[0].Tags["is_critical"] == "True";

def on_event_data_received_handler(stream: StreamConsumer, data: EventData):
    with data:
        print("Event read for stream. Event Id: " + data.Id)

stream_received.events.on_data_received = on_event_data_received_handler

newStream.Events.OnDataReceived += (stream, args) =>
{
    Console.WriteLine($"Event received for stream. Event Id: {args.Data.Id}");
};

from quixstreams import CommitOptions

commit_settings = CommitOptions()
commit_settings.commit_every = 100 # note, you can set this to None
commit_settings.commit_interval = 500 # note, you can set this to None
commit_settings.auto_commit_enabled = True
topic_consumer = client.get_topic_consumer('yourtopic', commit_settings=commit_settings)

var topicConsumer = client.GetTopicConsumer(topic, consumerGroup, new CommitOptions()
{
        CommitEvery = 100,
        CommitInterval = 500,
        AutoCommitEnabled = true // optional, defaults to true
});

from quixstreams import CommitMode

topic_consumer = client.get_topic_consumer('yourtopic', commit_settings=CommitMode.Manual)

client.GetTopicConsumer(topic, consumerGroup, CommitMode.Manual);

topic_consumer.commit()

topicConsumer.Commit();

topic_consumer = client.get_topic_consumer(test_topic, auto_offset_reset=AutoOffsetReset.Latest)
or
topic_consumer = client.get_topic_consumer(test_topic, auto_offset_reset=AutoOffsetReset.Earliest)

var topicConsumer = client.GetTopicConsumer("MyTopic", autoOffset: AutoOffsetReset.Latest);
or
var topicConsumer = client.GetTopicConsumer("MyTopic", autoOffset: AutoOffsetReset.Earliest);

def on_revoking_handler(topic_consumer: TopicConsumer):
    # your code

topic_consumer.on_revoking = on_revoking_handler

topicConsumer.OnRevoking += (sender, args) =>
    {
        // ... your code ...
    };

from quixstreams import StreamConsumer

def on_streams_revoked_handler(topic_consumer: TopicConsumer, streams: [StreamConsumer]):
    for stream in streams:
        print("Stream " + stream.stream_id + " got revoked")

topic_consumer.on_streams_revoked = on_streams_revoked_handler

topicConsumer.OnStreamsRevoked += (sender, revokedStreams) =>
    {
        // revoked streams are provided to the handler
    };

from quixstreams import *

client = KafkaStreamingClient('127.0.0.1:9092')

topic_consumer = client.get_topic_consumer(TOPIC_ID)

# read streams
def on_stream_received_handler(stream_received: StreamConsumer):    
    buffer = stream_received.timeseries.create_buffer()
    buffer.on_data_released = on_data_released_handler

def on_data_released_handler(stream: StreamConsumer, data: TimeseriesData):
    with data:
        df = data.to_dataframe()
        print(df.to_string())    

# Hook up events before initiating read to avoid losing out on any data
topic_consumer.on_stream_received = on_stream_received_handler

# Hook up to termination signal (for docker image) and CTRL-C
print("Listening to streams. Press CTRL-C to exit.")

# Handle graceful exit
App.run()

using System;
using System.Linq;
using System.Threading;
using QuixStreams.Streaming;
using QuixStreams.Streaming.Configuration;
using QuixStreams.Streaming.Models;


namespace ReadHelloWorld
{
    class Program
    {
        /// <summary>
        /// Main will be invoked when you run the application
        /// </summary>
        static void Main()
        {
            // Create a client which holds generic details for creating input and output topics
            var client = new KafkaStreamingClient("127.0.0.1:9092")

            using var topicConsumer = client.GetTopicConsumer(TOPIC_ID);

            // Hook up events before initiating read to avoid losing out on any data
            topicConsumer.OnStreamReceived += (topic, streamConsumer) =>
            {
                Console.WriteLine($"New stream read: {streamConsumer.StreamId}");

                var buffer = streamConsumer.Timeseries.CreateBuffer();

                buffer.OnDataReleased += (sender, args) =>
                {
                    Console.WriteLine($"ParameterA - {ags.Data.Timestamps[0].Timestamp}: {ags.Data.Timestamps.Average(a => a.Parameters["ParameterA"].NumericValue)}");
                };
            };

            Console.WriteLine("Listening for streams");

            // Hook up to termination signal (for docker image) and CTRL-C and open streams
            App.Run();

            Console.WriteLine("Exiting");
        }
    }
}

from quixstreams import RawTopicConsumer, RawMessage

raw_consumer = client.create_raw_topic_consumer(TOPIC_ID)

def on_message_received_handler(topic: RawTopicConsumer, msg: RawMessage):
    #bytearray containing bytes received from kafka
    data = msg.value

    #broker metadata as dict
    meta = msg.metadata

raw_consumer.on_message_received = on_message_received_handler
raw_consumer.subscribe()  # or use App.run()

var rawConsumer = client.GetRawTopicConsumer(TOPIC_ID)

rawConsumer.OnMessageRead += (sender, message) =>
{
    var data = (byte[])message.Value;
};

rawConsumer.Subscribe()  // or use App.Run()