Developing HPE Data Fabric Streams C#/.NET Applications

Before Your Begin

Confirm that your environment meets the following requirements:

Create a Producer Application

In the following example code, three messages are produced to a topic named mytopic in a stream named my_stream.

class Producer
    {
        public static async void Produce()
        {
            string stream = "/my_stream";
            string topicName = "mytopic";

            var config = new Dictionary<string, object> { { "streams.producer.default.stream", stream } };
            var messages = new string[] { "Msg1", "Msg2", "Msg3" };

            using (var producer = new Producer<Null, string>(config, null, new StringSerializer(Encoding.UTF8)))
            {
                foreach (var msg in messages)
                {
                    var deliveryReport = await producer.ProduceAsync(topicName, null, msg);
                    Console.WriteLine($"Delivery report:{deliveryReport.TopicPartitionOffset}");
                }

                producer.Flush(TimeSpan.FromSeconds(1));
            }
        }
    }

Create a Consumer Application

In following example code, the HPE Data Fabric Streams consumer is subscribed to my_stream/mytopic and it prints the content of each message that it reads.

using Confluent.Kafka;
using Confluent.Kafka.Serialization

    class Consumer
    {
        public static void Consume()
        {
            var stream = "/mystream";
            var topic = "mytopic";

            var config = new Dictionary<string, object>
            {
                { "group.id", "simple-csharp-consumer" },
                { "streams.consumer.default.stream", stream }
            };

            bool running = true;

            using (var consumer = new Consumer<Ignore, string>(config, null, new StringDeserializer(Encoding.UTF8)))
            {
                var l = new List<TopicPartitionOffset> { new TopicPartitionOffset(topic, 0, 0) };
                consumer.Assign(l);


                // Raised on critical errors, e.g. connection failures.
                consumer.OnError += (_, error) =>
                {
                    Console.WriteLine($"Error: {error}");
                    running = false;
                };

                // Raised on deserialization errors or when a consumed message has an error != NoError.
                consumer.OnConsumeError += (_, error) =>
                {
                    Console.WriteLine($"Consume error: {error}");
                    running = false;
                };

                while (running)
                {
                    Message<Ignore, string> msg;
                    if (consumer.Consume(out msg, TimeSpan.FromSeconds(10)))
                    {
                        Console.WriteLine($"Topic: {msg.Topic} Partition: {msg.Partition} Offset: {msg.Offset} {msg.Value}");
                    }
                }
            }
        }
    }

Run the Example Applications

To run the sample producer and consumer applications: