Lib.ApacheKafka-ProtoBuf.CLient 1.0.0

IKafkaProtobufClient

Thu vien nay cung cap mot API don gian de publish/consume message Protobuf qua Apache Kafka.

1) Cai dat package

Project da khai bao cac package can thiet:

• Confluent.Kafka
• Google.Protobuf
• Grpc.Tools (chi dung de sinh code tu .proto)

2) Ket noi

using Lib.ApacheKafka_ProtoBuf.CLient;
using Lib.ApacheKafka_ProtoBuf.CLient.Abstractions;

var client = IKafkaProtobufClient.Connect(new KafkaProtobufClientOptions
{
    Host = "localhost:9092",
    ClientId = "service-a"
});

Neu ket noi that bai, Connect(...) se nem Exception.

3) Publish message Protobuf

// Gui message protobuf thong thuong
client.PublishTopic("account.created", myProtoMessage);

// Gui message protobuf co id (request/response)
client.PublishTopicWithId("account.request", 1001, myProtoMessage);

4) Consume message Protobuf

// Co GroupName: trong cung 1 group thi chi 1 consumer nhan duoc message
client.ConsumerTopic<MyProto>(
    "account.created",
    "account-worker-group",
    data =>
    {
        Console.WriteLine(data);
    });

// Khong truyen GroupName: moi client se nhan topic (broadcast theo group rieng)
client.ConsumerTopic(
    "account.created",
    new MyProto(),
    data =>
    {
        Console.WriteLine(data);
    });

5) Consume theo id (request/response)

client.ConsumerTopicWithId<MyProto>(
    topicName: "account.request",
    id: 1001,
    groupName: "account-response-group",
    callback: data =>
    {
        Console.WriteLine($"Success: {data}");
    },
    timeoutMs: 5000,
    error: ex =>
    {
        Console.WriteLine($"Error: {ex.Message}");
    });

// Overload khong truyen group
client.ConsumerTopicWithId<MyProto>(
    topicName: "account.request",
    id: 1002,
    callback: data => Console.WriteLine(data),
    timeoutMs: 5000,
    error: ex => Console.WriteLine(ex.Message));

Neu qua timeoutMs ma khong co response, callback error se nhan TimeoutException.

6) Day loi cho luong ConsumerTopicWithId

client.ThrowException("account.request", 1001, "Du lieu khong hop le");

Message loi se duoc gui duoi dang wrapper type = 0, va ben dang doi ConsumerTopicWithId se nhan qua callback error.

7) Ngat ket noi

client.Disconnect();
// hoac
client.Dispose();

8) Lenh push len NuGet

Cach 1: Push truc tiep bang dotnet nuget push

$env:NUGET_API_KEY = Read-Host "Nhap NuGet API key" -MaskInput
dotnet pack "E:\Ninh\CSharp\BackendSimpleWar\Lib.ApacheKafka-ProtoBuf.CLient\Lib.ApacheKafka-ProtoBuf.CLient.csproj" -c Release -o "E:\Ninh\CSharp\BackendSimpleWar\artifacts\nuget"
dotnet nuget push "E:\Ninh\CSharp\BackendSimpleWar\artifacts\nuget\Lib.ApacheKafka-ProtoBuf.CLient.1.0.0.nupkg" --source "https://api.nuget.org/v3/index.json" --api-key $env:NUGET_API_KEY --skip-duplicate

Cach 2: Dung script tu dong tang version + push

$env:NUGET_API_KEY = Read-Host "Nhap NuGet API key" -MaskInput
powershell -ExecutionPolicy Bypass -File "E:\Ninh\CSharp\BackendSimpleWar\tools\scripts\publish-nuget.ps1" -BaseVersion "1.0.1"

Neu chi muon pack ma khong push:

powershell -ExecutionPolicy Bypass -File "E:\Ninh\CSharp\BackendSimpleWar\tools\scripts\publish-nuget.ps1" -BaseVersion "1.0.1" -SkipPush