PubSub example

[AllanMichaelsen]

Hey
I was wondering if anyone has an example for adding PubSub functionality with Servicemodel?

Running the services isn't much of an issue (great project), but I'm struggling a little with the PubSub modeling.

I hope someone can throw me a bone (example implementation)

[max-ieremenko]

Hi,
more information would be beneficial. What is PubSub functionality and what do you want to implement?

[AllanMichaelsen]

Yes of course. Sorry for the rather short request.

I have a order service that was implemented with SignalR, but the desire is to migrate it to gRPC.

So the client creates a connection to the services for Order & OrderNotification

Client subscribes to either create and/or update order notifications (via the OrderNotification endpoint)
The endpoint should:

• Have queues for events of different types (create/update)
• Keep track up individual clients that has subscribed to events and broadcast notifications to those clients when an event is placed in the corresponding queue.

Here's a couple of usage scenarios:

1. Create an order

• Create order is called (via order endpoint)
• A notification is placed in the create queue
• The notification is dequeued and sent out to all clients that has subscribed to create notifications (as stated above)

2. Update an order

• Update order is called (via order endpoint)
• A notification is placed in the update queue
• The notification is dequeued and sent out to all clients that has subscribed to update notifications (as stated above)

So I'm thinking a long-lived connection returning a IAsyncEnumerable that on the client side yield the results as it is received from the server.

But should the server-side service be Scoped/Singleton (I'm thinking it shouldn't be Transient)?
Any special considerations I should think of Client-side?

I hope that clarifies what I'm thinking.

[AllanMichaelsen]

Maybe something in the style of:

internal class OrderService : IOrderService
{
    private readonly Queue<OrderDto> orderCreateQueue = new();

    public void OrderCreate(OrderDto dto)
    {
        orderCreateQueue.Enqueue(dto);
    }

    public async IAsyncEnumerable<OrderDto> OrderCreateSubscribe(ServerCallContext context)
    {
        while (!context.CancellationToken.IsCancellationRequested)
        {
            while (orderCreateQueue.Any())
            {
                var queueObject = orderCreateQueue.Dequeue();
                yield return queueObject;
            }

            await Task.Delay(100);
        }
    }

Only thing is with this I dont get messages to ALL recipients. Just one.

In SignalR it would be:

` public async Task Upsert(OrderDto dto)
{
...
Save data
...

    await Clients
        .AllExcept(Context.ConnectionId)
        .Upsert(dto);

    return dto;
}

`

Suggestions?

[AllanMichaelsen]

There are plenty of examples for a "chat" application that does PubSub.

https://github.com/chgc/grpc-dotnetcore-3-chat

But those use protobuf which I'd like to avoid.

[AllanMichaelsen]

I have a solution with Server streaming and a ReactiveX stream. Works flawless.