using RabbitMQ.Client;
using System.Text;
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: "RabbitMqDotNetTutorial.01",
durable: false,
exclusive: false,
autoDelete: false,
arguments: null);
while (true)
{
Console.WriteLine(AppInfo.Value);
Console.WriteLine("Write what you want to send");
Console.WriteLine("Write nothing to exit.");
string usermessage = Console.ReadLine();
if (string.IsNullOrWhiteSpace(usermessage))
break;
string message = usermessage;
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: "",
routingKey: "RabbitMqDotNetTutorial.01",
basicProperties: null, body);
Console.WriteLine("");
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine("\tSent {0}", message);
Console.ForegroundColor = ConsoleColor.Gray;
Console.WriteLine("");
}
}
using RabbitMQ.Client.Events;
using RabbitMQ.Client;
using System.Text;
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: "RabbitMqDotNet6Tutorial.01",
durable: false,
exclusive: false, autoDelete: false, arguments: null);
Console.WriteLine(AppInfo.Value);
Console.WriteLine(" Waiting for messages.");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var message = Encoding.UTF8.GetString(ea.Body.ToArray());
Console.Write("-> Received: ");
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine(message);
Console.ForegroundColor = ConsoleColor.Gray;
//channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
};
channel.BasicConsume(queue: "RabbitMqDotNet6Tutorial.01",
autoAck: true, consumer: consumer);
char key = 'z';
while (key != 'q' && key != 'Q')
{
Console.WriteLine(" Press [q] or [Q] to exit.");
key = Console.ReadKey().KeyChar;
}
}
using EasyNetQ;
using (var bus = RabbitHutch.CreateBus
("host=localhost"))
{
for (int i = 0; i < 10; i++)
{
await bus.SendReceive.SendAsync
("1EasyNetMQ.Queue",
new TextMessage()
{
Text = i + ": Hello World from EasyNetQ"
}
);
}
}
using EasyNetQ;
using (var bus = RabbitHutch.CreateBus("host=localhost"))
{
await bus.SendReceive.ReceiveAsync("1EasyNetMQ.Queue",
HandleTextMessage);
Console.WriteLine("Listening for messages. Hit to quit.");
Console.ReadLine();
}
static void HandleTextMessage(TextMessage textMessage)
{
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine("Got message: {0}", textMessage.Text);
Console.ResetColor();
}
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
//durbale = true ponieważ chcemy aby nasze zdania istniały po restarcie
channel.QueueDeclare(queue: "RabbitMqDotNet6Tutorial.02", true,
false, false, null);
while (true)
{
Console.WriteLine("Write [.] to increase time to do job for worker. Foreach [.]");
Console.WriteLine("Write [!] at least once. To make one worker fail job");
Console.WriteLine("Write [q] or [Q] to exit.");
string whatuserwrote = Console.ReadLine();
if (whatuserwrote == "q" || whatuserwrote == "Q")
break;
if (string.IsNullOrEmpty(whatuserwrote))
continue;
Job job = CreateJob(whatuserwrote);
string message = JsonConvert.SerializeObject(job);
var body = Encoding.UTF8.GetBytes(message);
var properties = channel.CreateBasicProperties();
properties.Persistent = true;
channel.BasicPublish("", routingKey: "RabbitMqDotNet6Tutorial.02",
properties, body);
WriteMessageOnConsole(message);
}
}
void WriteMessageOnConsole(string message)
{
Console.WriteLine("");
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine("\tSent {0}", message);
Console.ForegroundColor = ConsoleColor.Gray;
Console.WriteLine("");
}
Job CreateJob(string usertext)
{
int howManySecondsWillJobTake = usertext.Split('.').Length - 1;
bool shouldFail = usertext.IndexOf('!') > 0;
string message = usertext.Replace("!", "")
.Replace(".", "");
return new Job()
{
Message = message,
HowManySecondsWillJobTake = howManySecondsWillJobTake,
Type = JobType.SendEmail,
ShouldFaillOnWorkerTwo = shouldFail
};
}
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;
using System.Text.Json;
Console.ForegroundColor = GetRandomConsoleColor();
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: "RabbitMqDotNet6Tutorial.02",
durable: true,
exclusive: false, autoDelete: false, arguments: null);
channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
Console.WriteLine(" [*] Waiting for jobs.");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
byte[] body = ea.Body.ToArray();
var jobAsJsonText = Encoding.UTF8.GetString(body);
Job job = JsonSerializer.Deserialize<Job>(jobAsJsonText);
Console.WriteLine(" [>] Received {0}", job.Message);
Console.WriteLine(" [>] Received {0}", job.Type);
Thread.Sleep(job.HowManySecondsWillJobTake * 1000);
channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine(" [>] Done");
Console.ForegroundColor = ConsoleColor.Gray;
};
channel.BasicConsume(queue: "RabbitMqDotNet6Tutorial.02",
autoAck: false,
consumer: consumer);
Console.WriteLine(" Press [enter] to end program.");
Console.ReadLine();
}
ConsoleColor GetRandomConsoleColor()
{
Random _random = new Random();
var consoleColors = Enum.GetValues(typeof(ConsoleColor));
return (ConsoleColor)consoleColors.
GetValue(_random.Next(consoleColors.Length));
}
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;
using System.Text.Json;
Console.ForegroundColor = GetRandomConsoleColor();
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
while (true)
{
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: "RabbitMqDotNet6Tutorial.02",
durable: true, exclusive: false,
autoDelete: false, arguments: null);
channel.BasicQos(prefetchSize: 0,
prefetchCount: 1,
global: false);
Console.WriteLine(" [*] Waiting for jobs.");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
byte[] body = ea.Body.ToArray();
var jobAsJsonText = Encoding.UTF8.GetString(body);
Job job = JsonSerializer.Deserialize<Job>(jobAsJsonText);
Console.WriteLine(" [>] Received {0}", job.Message);
Console.WriteLine(" [>] Received {0}", job.Type);
if (job.ShouldFaillOnWorkerTwo == false)
{
Thread.Sleep(job.HowManySecondsWillJobTake * 500);
channel.BasicAck(deliveryTag: ea.DeliveryTag,
multiple: false);
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine(" [>] Done");
Console.ForegroundColor = ConsoleColor.Gray;
}
else
{
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine($" [x] Fail {job.Message}");
Console.WriteLine($" [x] Fail {job.Type}");
Console.ForegroundColor = ConsoleColor.Gray;
}
};
channel.BasicConsume
(queue: "RabbitMqDotNet6Tutorial.02",
autoAck: false,
consumer: consumer);
Console.WriteLine
(" Press [enter] to restart connection.CreateModel()");
Console.ReadLine();
}
}
ConsoleColor GetRandomConsoleColor()
{
Random _random = new Random();
var consoleColors = Enum.GetValues(typeof(ConsoleColor));
return (ConsoleColor)consoleColors.GetValue(_random.Next(consoleColors.Length));
}
using EasyNetQ;
using Newtonsoft.Json;
while (true)
{
Console.WriteLine
("Write [.] to increase time to do job for worker. Foreach [.]");
Console.WriteLine(
"Write [!] at least once. To make one worker fail job");
Console.WriteLine
("Write [q] or [Q] to exit.");
string whatuserwrote = Console.ReadLine();
if (whatuserwrote == "q" || whatuserwrote == "Q")
break;
if (string.IsNullOrEmpty(whatuserwrote))
continue;
Job job = CreateJob(whatuserwrote);
using (var bus = RabbitHutch.CreateBus("host=localhost"))
{
await bus.SendReceive.SendAsync<Job>("2EasyNetMQ.NewJob.Queue"
, job
);
}
string message = JsonConvert.SerializeObject(job);
WriteMessageOnConsole(message);
}
Job CreateJob(string usertext)
{
int howManySecondsWillJobTake = usertext.Split('.').Length - 1;
bool shouldFail = usertext.IndexOf('!') > 0;
string message = usertext.Replace("!", "")
.Replace(".", "");
return new Job()
{
Message = message,
HowManySecondsWillJobTake = howManySecondsWillJobTake,
Type = JobType.SendEmail,
ShouldFaillOnWorkerTwo = shouldFail
};
}
void WriteMessageOnConsole(string message)
{
Console.WriteLine("");
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine("\tSent {0}", message);
Console.ForegroundColor = ConsoleColor.Gray;
Console.WriteLine("");
}
using EasyNetQ;
var connectionString = "host=localhost";
// RabbitMQ connection string
using (var bus = RabbitHutch.CreateBus(connectionString))
{
var queueName = "2EasyNetMQ.NewJob.Queue";
// Replace with your queue name
bus.Advanced.QueueDeclare(queueName, durable: true, false, false);
await bus.SendReceive.ReceiveAsync
(queueName, HandleMessage, x => x.WithAutoDelete(false));
Console.WriteLine("Press [enter] to exit.");
Console.ReadLine();
}
static void HandleMessage(Job job)
{
try
{
// Process the message
Console.WriteLine(" [>] Received {0}", job.Message);
Console.WriteLine(" [>] Received {0}", job.Type);
Thread.Sleep(job.HowManySecondsWillJobTake * 1000);
// Manually ACK the message
// YOU CAN'T DO THIS WITH EasyNetQ
// Note: EasyNetQ automatically ACKs messages by default.
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine(" [>] Done");
Console.ForegroundColor = ConsoleColor.Gray;
}
catch (Exception ex)
{
Console.WriteLine($"Error processing message: {ex.Message}");
// Handle any exceptions or log errors here
}
}
using EasyNetQ;
var connectionString = "host=localhost";
// RabbitMQ connection string
using (var bus = RabbitHutch.CreateBus(connectionString))
{
var queueName = "2EasyNetMQ.NewJob.Queue";
// Replace with your queue name
bus.Advanced.QueueDeclare(queueName, durable: true, false, false);
// Subscribe to messages
await bus.SendReceive.ReceiveAsync
(queueName, HandleMessage, x => x.WithAutoDelete(false));
Console.WriteLine("Press [enter] to exit.");
Console.ReadLine();
}
static void HandleMessage(Job job)
{
if (job.ShouldFaillOnWorkerTwo == false)
{
// Note: EasyNetQ automatically ACKs messages by default.
Thread.Sleep(job.HowManySecondsWillJobTake * 500);
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine(" [>] Done");
Console.ForegroundColor = ConsoleColor.Gray;
}
else
{
// Manually N-ACK the message
// YOU CAN'T DO THIS WITH EasyNetQ
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine($" [x] Fail {job.Message}");
Console.WriteLine($" [x] Fail {job.Type}");
Console.ForegroundColor = ConsoleColor.Gray;
// THIS WILL SEND MESSAGE TO DEAD-LETTER QUEQUE
// DEFAULT NAME OF THIS QUEQUE IS
// EasyNetQ_Default_Error_Queue
throw new Exception("JOB FAILED");
}
}
using RabbitMQ.Client;
using System.Text;
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.ExchangeDeclare(exchange: "RabbitMqDotNet6Tutorial.03",
type: ExchangeType.Fanout,
durable: true, autoDelete: false, arguments: null);
while (true)
{
var message = "Daj mi ocene 5!";
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: "RabbitMqDotNet6Tutorial.03",
routingKey: "",
basicProperties: null, body: body);
Console.WriteLine(" [x] Sent {0}", message);
Console.ReadLine();
}
}
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.ExchangeDeclare(exchange: "RabbitMqDotNet6Tutorial.03",
type: ExchangeType.Fanout, durable: true, autoDelete: false, arguments: null);
var queueName = channel.QueueDeclare().QueueName;
channel.QueueBind(queue: queueName,
exchange:
"RabbitMqDotNet6Tutorial.03", routingKey: "");
Console.WriteLine(" [*] Waiting for logs.");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
byte[] body = ea.Body.ToArray();
var message = Encoding.UTF8.GetString(body);
Console.WriteLine(" [x] {0}", message);
};
channel.BasicConsume(queue: queueName, autoAck: true, consumer: consumer);
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
using EasyNetQ;
using EasyNetQ.Topology;
using (var bus = RabbitHutch.CreateBus("host=localhost"))
{
var message = "Daj mi ocene 5!";
await bus.PubSub.PublishAsync<String>(
message
);
Console.WriteLine(" [x] Sent {0}", message);
Console.ReadLine();
}
static void AddCustom(IBus bus)
{
var exchange = bus.Advanced.ExchangeDeclare(
"3EasyNetQ.PublishToMany.Exchange", ExchangeType.Fanout);
var queue = bus.Advanced.QueueDeclare
("3EasyNetQ.PublishToMany.Queue",
durable: true, exclusive: false, autoDelete: true);
bus.Advanced.Bind(exchange, queue, "#");
}
using EasyNetQ;
using EasyNetQ.Topology;
using (var bus = RabbitHutch.CreateBus("host=localhost"))
{
AddCustom(bus);
await bus.PubSub.SubscribeAsync("3EasyNetQ.SubscribeSecond", HandleTextMessage);
Console.WriteLine("Listening for messages. Hit to quit.");
Console.ReadLine();
}
static void HandleTextMessage(string textMessage)
{
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine("Got message: {0}", textMessage);
Console.ResetColor();
}
static void AddCustom(IBus bus)
{
var exchange = bus.Advanced.ExchangeDeclare(
"3EasyNetQ.PublishToMany.Exchange", ExchangeType.Fanout);
var queue = bus.Advanced.QueueDeclare
("3EasyNetQ.PublishToMany.Queue",
durable: true, exclusive: false, autoDelete: true);
bus.Advanced.Bind(exchange, queue, "#");
}
public class DocumentSavedEvent
{
private Guid id;
private string filename;
[JsonConstructor]
public DocumentSavedEvent(Guid id, string filename)
{
this.Id = id;
this.FileName = filename;
}
public Guid Id { get => id; set => id = value; }
public string FileName { get => filename; set => filename = value; }
}
public class TradeRecordedEvent
{
private Guid id;
private string commodity;
private int quantity;
[JsonConstructor]
public TradeRecordedEvent(Guid id, string commodity, int quantity)
{
this.Id = id;
this.Commodity = commodity;
this.Quantity = quantity;
}
public Guid Id
{ get => id; set => id = value; }
public string Commodity
{ get => commodity; set => commodity = value; }
public int Quantity
{ get => quantity; set => quantity = value; }
}
public class UserLoggedEvent
{
private Guid id;
private string username;
[JsonConstructor]
public UserLoggedEvent(Guid id, string username)
{
this.Id = id;
this.UserName = username;
}
public Guid Id { get => id; set => id = value; }
public string UserName { get => username; set => username = value; }
}
using Rebus.Activation;
using Rebus.Bus;
using Rebus.Config;
using (var activator = new BuiltinHandlerActivator())
{
var bus = Configure.With(activator)
.Transport(t => t.UseRabbitMqAsOneWayClient
("amqp://guest:guest@localhost:5672"))
.Start();
await Task.Delay(500);
Console.WriteLine("Trading is running");
while (true)
{
Console.WriteLine(@"
a) Publish TradeRecordedEvent
b) Publish DocumentSavedEvent
c) Publish UserLogedEvent
q) Quit");
var keyChar = char.ToLower(Console.ReadKey(true).KeyChar);
switch (keyChar)
{
case 'a':
await SendTradeRecordedEventAsync(bus);
break;
case 'b':
await SendDocumentSavedAsync(bus);
break;
case 'c':
await SendUserLoggedEventAsync(bus);
break;
case 'q':
break;
default:
Console.WriteLine("There's no option ({0})", keyChar);
break;
}
}
}
async Task SendTradeRecordedEventAsync(IBus bus)
{
Console.WriteLine("Please enter new trade details");
Console.WriteLine(" commodity > ");
var commodity = Console.ReadLine();
if (string.IsNullOrWhiteSpace(commodity)) return;
int quantity;
Console.WriteLine(" quantity > ");
while (!int.TryParse(Console.ReadLine(), out quantity)) ;
var tradeEvent = new TradeRecordedEvent
(Guid.NewGuid(), commodity, quantity);
await bus.Publish(tradeEvent);
Console.WriteLine("Sended tradeEvent");
}
async Task SendDocumentSavedAsync(IBus bus)
{
Console.WriteLine("Please enter new document details");
Console.WriteLine(" document name > ");
var documentName = Console.ReadLine();
if (string.IsNullOrWhiteSpace(documentName)) return;
var docEvent = new DocumentSavedEvent
(Guid.NewGuid(), documentName);
await bus.Publish(docEvent);
Console.WriteLine("Sended docEvent");
}
async Task SendUserLoggedEventAsync(IBus bus)
{
Console.WriteLine("Please enter new trade details");
Console.WriteLine(" Username > ");
var username = Console.ReadLine();
if (string.IsNullOrWhiteSpace(username)) return;
var userEvent = new UserLoggedEvent(Guid.NewGuid(), username);
await bus.Publish(userEvent);
Console.WriteLine("Sended userEvent");
}
using Rebus.Activation;
using Rebus.Config;
using Rebus.Routing.TypeBased;
using (var activator = new BuiltinHandlerActivator())
{
activator.Register(() => new TradeRecordedEventHandler());
activator.Register(() => new UserLoggedEventHandler());
activator.Register(() => new DocumentSavedEventHandler());
var bus = Configure.With(activator)
.Logging
(l => l.ColoredConsole(minLevel: Rebus.Logging.LogLevel.Warn))
.Transport
(t => t.UseRabbitMq("amqp://guest:guest@localhost:5672",
"Rebus.SystemTooBig.Trading"))
.Routing
(r => r.TypeBased().
MapAssemblyOf<TradeRecordedEvent>("Rebus.SystemTooBig.Trading"))
.Start();
await bus.Subscribe<TradeRecordedEvent>();
await bus.Subscribe<UserLoggedEvent>();
await bus.Subscribe<DocumentSavedEvent>();
Console.WriteLine("Invoicing is running - press Enter to quite");
Console.ReadKey();
}
internal class DocumentSavedEventHandler : IHandleMessages<DocumentSavedEvent>
{
public DocumentSavedEventHandler()
{
}
public async Task Handle(DocumentSavedEvent message)
{
Console.WriteLine
($"Document was saved: {message.Id} ({message.FileName})");
}
}
internal class TradeRecordedEventHandler : IHandleMessages<TradeRecordedEvent>
{
public TradeRecordedEventHandler()
{
}
public async Task Handle(TradeRecordedEvent message)
{
Console.WriteLine
($"Invocing trade: {message.Id} ({message.Quantity} x {message.Commodity})");
}
}
internal class UserLoggedEventHandler : IHandleMessages<UserLoggedEvent>
{
public UserLoggedEventHandler()
{
}
public async Task Handle(UserLoggedEvent message)
{
Console.WriteLine
($"User was logged: {message.Id} ({message.UserName})");
}
}
using _1Kafka.Confluent.Common;
using Confluent.Kafka;
using System.Net;
using System.Text.Json;
string bootstrapServers = "localhost:9092";
string topic = "1Kafka.Confluent.Test";
char c = ' ';
int messageId = 0;
while (c != 'q')
{
var messagerequest = GetMessageRequest();
string json = JsonSerializer.Serialize(messagerequest);
await SendMessageRequest(topic, json);
Console.WriteLine("Press any key to continue");
Console.WriteLine("Press 'q' to exit");
c = Console.ReadKey().KeyChar;
Console.WriteLine("");
}
MessageRequest GetMessageRequest()
{
string message = "";
int color = -1;
while (string.IsNullOrWhiteSpace(message) || color == -1)
{
Console.WriteLine("Write Your Message");
message = Console.ReadLine();
Console.WriteLine("Choose Color");
for (int i = 0; i < 16; i++)
{
Console.ForegroundColor = (ConsoleColor)i;
Console.WriteLine($"{i} . {((ConsoleColor)i).ToString()}");
}
Console.ResetColor();
int colora;
bool check = int.TryParse(Console.ReadLine(), out colora);
if (!check)
continue;
color = colora;
}
messageId++;
MessageRequest messageRequest = new()
{
Color = color,
Message = message,
Id = messageId,
UniqueId = Guid.NewGuid(),
StartProducer = DateTime.Now
};
return messageRequest;
}
async Task<bool> SendMessageRequest(string topic, string message)
{
ProducerConfig config = new ProducerConfig
{
BootstrapServers = bootstrapServers,
ClientId = Dns.GetHostName(),
};
try
{
using (var producer = new ProducerBuilder
<Null, string>(config).Build())
{
var result = await producer.ProduceAsync
(topic, new Message<Null, string>
{
Value = message
});
Console.WriteLine($"Delivery Timestamp:{result.Timestamp.UtcDateTime}");
return await Task.FromResult(true);
}
}
catch (Exception ex)
{
Console.WriteLine($"Error occured: {ex.Message}");
}
return await Task.FromResult(false);
}
using _1Kafka.Confluent.Common;
using Confluent.Kafka;
using System.Net;
using System.Text.Json;
int unique = 5;
string topic = "1Kafka.Confluent.Test";
string groupId = "1Kafka.Confluent.Test_Group_" + unique.ToString();
string bootstrapServers = "localhost:9092";
Console.WriteLine($"{topic} GROUP ID : {groupId}");
var config = new ConsumerConfig
{
GroupId = groupId,
BootstrapServers = bootstrapServers,
AutoOffsetReset = AutoOffsetReset.Earliest
};
try
{
using (var consumerBuilder = new ConsumerBuilder
<Ignore, string>(config).Build())
{
consumerBuilder.Subscribe(topic);
var cancelToken = new CancellationTokenSource();
try
{
while (true)
{
var consumer = consumerBuilder.Consume
(cancelToken.Token);
var messageRequest = JsonSerializer.Deserialize
<MessageRequest>
(consumer.Message.Value);
Console.WriteLine($"\nProducerStarted: {messageRequest.StartProducer}");
Console.WriteLine($"\nProcessing Number: {messageRequest.Id}");
Console.WriteLine($"Processing UniqueId: {messageRequest.UniqueId}");
if (messageRequest.Color >= 0 && messageRequest.Color <= 15)
Console.ForegroundColor = (ConsoleColor)messageRequest.Color;
Console.WriteLine(messageRequest.Message);
Console.ResetColor();
Console.WriteLine("");
}
}
catch (OperationCanceledException)
{
consumerBuilder.Close();
}
}
}
catch (Exception ex)
{
Console.WriteLine(ex.Message);
}
using KafkaFlow;
using KafkaFlow.Producers;
using KafkaFlow.Serializer;
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddKafka(
kafka => kafka
.AddCluster(cluster =>
{
const string topicName = "1KafkaFlow";
cluster
.WithBrokers(new[] { "localhost:9092" })
.CreateTopicIfNotExists(topicName,
numberOfPartitions: 3, replicationFactor: 3)
.AddProducer(
name: "1KafkaFlow.Producer",
producer => producer
.DefaultTopic(topicName)
.AddMiddlewares(middlewares =>
middlewares
.AddSerializer<JsonCoreSerializer>()));
})
);
var app = builder.Build();
app.MapPost("/add", RequestHandler.HandleAsync);
app.Run();
// Handler
public static class RequestHandler
{
public static async Task<IResult> HandleAsync(
IProducerAccessor producerAccessor,
AddTaskRequest request, CancellationToken cancellationToken)
{
var producer = producerAccessor.GetProducer("1KafkaFlow.Producer");
await producer.ProduceAsync(
null,
request
);
return Results.Accepted();
}
}
using KafkaFlow;
using KafkaFlow.Serializer;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
const string topicName = "1KafkaFlow";
int unique = 5;
string groupId = "1KafkaFlow_Group_" + unique.ToString();
var services = new ServiceCollection();
services.AddKafkaFlowHostedService(
kafka => kafka
.UseMicrosoftLog()
.AddCluster(cluster =>
{
cluster
.WithBrokers(new[] { "localhost:9092" })
.AddConsumer(consumer =>
consumer
.Topic(topicName)
.WithGroupId(groupId)
.WithBufferSize(100)
.WithWorkersCount(3)
.WithAutoOffsetReset(KafkaFlow.AutoOffsetReset.Earliest)
.AddMiddlewares(middlewares => middlewares
.AddDeserializer<JsonCoreDeserializer>()
.AddTypedHandlers(handlers =>
handlers.AddHandler<AddTaskHandler>()
)
)
);
})
);
services.AddLogging(configure => configure.AddConsole());
var provider = services.BuildServiceProvider();
var bus = provider.CreateKafkaBus();
await bus.StartAsync();
Console.WriteLine("Press key to exit");
Console.ReadKey();
