Changing data model | allowing multiple messages to be sent via producer

This commit is contained in:
Brandon Watson 2024-06-14 11:08:48 -05:00
parent d9866c2fa9
commit 4570c43453
3 changed files with 45 additions and 25 deletions

View File

@ -38,7 +38,7 @@ class Consumer
var cr = consumer.Consume(cts.Token); var cr = consumer.Consume(cts.Token);
dataRepository.Save(cr.Message.Value); dataRepository.Save(cr.Message.Value);
Console.WriteLine($"Consumed message with id '{cr.Message.Value.id}' at: '{cr.TopicPartitionOffset}'."); Console.WriteLine($"Consumed message with id '{cr.Message.Value.id}'. Saving value to database");
} }
catch (ConsumeException e) catch (ConsumeException e)
{ {

View File

@ -7,35 +7,59 @@ using Model;
class Producer class Producer
{ {
public static string TOPIC_NAME = "test-topic";
public static async Task Main (string[] args) public static async Task Main (string[] args)
{ {
var config = new ProducerConfig { BootstrapServers = "localhost:29092" }; var config = new ProducerConfig { BootstrapServers = "localhost:29092" };
using (var producerBuilder = new ProducerBuilder<string, DataModel>(config) string introText = "\nType a string of text then press Enter. Type '+' anywhere in the text to quit:\n";
var input = "";
using (var producer = new ProducerBuilder<string, DataModel>(config)
.SetValueSerializer(new JsonSerializer<DataModel>()) .SetValueSerializer(new JsonSerializer<DataModel>())
.Build()) .Build())
{ {
try
{ Console.WriteLine(introText);
var msg = new Message<string, DataModel> do
{ {
Key = System.Guid.NewGuid().ToString(), input = Console.ReadLine();
Value = new DataModel(System.Guid.NewGuid().ToString()) try
{
if (input != String.Empty)
{ {
name = "Name", SendMessage(producer, input);
description = "Description",
notes = "This is a test object"
} }
};
}
var dr = await producerBuilder.ProduceAsync("test-topic", msg); catch (OverflowException e)
Console.WriteLine($"Delivered '{dr.Value}' to '{dr.TopicPartitionOffset}'"); {
} Console.WriteLine("An error has occurred", e.Message, input);
catch (ProduceException<string, DataModel> e) Console.WriteLine(introText);
{ }
Console.WriteLine($"Delivery failed: {e.Error.Reason}"); } while (input != "+");
} }
}
private static async void SendMessage(IProducer<String, DataModel> producer, string str)
{
try
{
var msg = new Message<string, DataModel>
{
Key = System.Guid.NewGuid().ToString(),
Value = new DataModel(System.Guid.NewGuid().ToString())
{
message = str
}
};
var dr = await producer.ProduceAsync(TOPIC_NAME, msg);
Console.WriteLine($"Delivered message with id '{dr.Value.id}'");
}
catch (ProduceException<string, DataModel> e)
{
Console.WriteLine($"Delivery failed: {e.Error.Reason}");
} }
} }
} }

View File

@ -9,10 +9,6 @@ namespace Model
public string id { get; set; } public string id { get; set; }
public string? name { get; set; } public string? message { get; set; }
public string? description { get; set; }
public string? notes { get; set; }
} }
} }