upta / pubsub Goto Github PK

Was wondering.

Hi,
Looks like a nice clean library, good job!
Can you tell me what scope this works on, is it possible to use this nuget package to communicate across a network(ethernet)? Or is it only intended for single system/pc use?
Thanks!

Not sure if it's an issue. Suppose I have 2 classes which I use to communicate: base and derived. Listener that is subscribed to the base class with a handler method gets called when an instance of the derived class is published.

why not make a peer to peer backend to have asynchronous messages across machines?

Could you please create a wiki?

I had to look in source code to discover the Hub class and find out its methods.

I'm subscribing in the constructor of a view, since many instances are closed and reopened, I thought unsubscribing first would work, but it doesn't.
Passing this doesn't work, but I thought that made sense, since the View I'm calling this from is a different instance. But I'm expecting the call without this to unsubscribe all.
Can anyone explain what I'm missing?

Hub.Default.Unsubscribe<RequestEmailEvent>();   // this is not unsubscribing from anything the next line creates, so I get compounding/multiple events, confused why
Hub.Default.Subscribe<RequestEmailEvent>(this, this.OnRequestEmail);

Is there a way to know if I have already subscribed to an event so that I do not subscribe to it again? I find that in certain circumstances, I have the event handlers firing multiple times when an event gets published because I have multiple duplicate subscriptions.

Can you add a method on the Hub class to clear the handler list?

I'm expecting events behavior from PubSub, i.e. if I publish an event, somebody will catch it eventually (or not) and do what needs to be done. I was really surprised that it waits for all handlers to finish, i.e. Publish won't return until all handlers are done.

PubSub.Hub.Default.Subscribe<MyObj>(async c =>
{
    await Task.Delay(TimeSpan.FromSeconds(10));
});

await PubSub.Hub.Default.PublishAsync(new MyObj());
var t = 1; // We reach this line only after the delay func is finished, i.e. in 10 seconds.

I would excpect that the above code behaves like this:

PubSub.Hub.Default.Subscribe<MyObj>(c =>
            {
                _ = Task.Run(async () => await Task.Delay(TimeSpan.FromSeconds(10)));
            });

await PubSub.Hub.Default.PublishAsync(new MyObj());
var t = 1; // We reach this line immideatelly. Handler task will be finished at some point in time.

I have a piece of code that takes an "event" and stores that event to CosmosDb.
I've verified that there is only a Singleton instance of my EventManagementService (constructor only runs once), and also that the publisher of the event only transmits the event once.
Here is the subscriber piece of code:

public void Initialize()
        {            
            _hub.Subscribe<NoobEvent>(SaveMessageToEventStore);            
        }

        private async void SaveMessageToEventStore(NbtEvent updateEvent)
        {
            _hub.UnSubscribe<NoobEvent>();

            _logger.Information($"Transmitting event to eventstore: {updateEvent.EventName} {updateEvent.EventType}");
            
            //TODO: (Pedro) Validate the updateEvent

            await _exceptionHandler.RunAsync(() => _eventStore.Write(updateEvent));

            _hub.Subscribe<NoobEvent>(SaveMessageToEventStore);
        }

If I do not unsubscribe while executing the "SaveMessageToEventstore()", then I see that it executes twice. Questions to this:

• Why does the subscriber invoke twice? I only have a single publisher, and a single handler
• By Unsubscribing at the beginning of the handler, does this run the risk of loosing any potentially queued messages, or will it resume where left off?
• Is there a way to mark a published message as "completed" that I am missing?

In advance, thanks.

Extending object is bad form and creates a messy developer experience in any library. Imagine using a graphics library and working in your view-model, or worse: your model! Intellisense would always be populated with graphics-related methods and properties like draw, rotate, x and y coordinates. Additionally, what if a user wanted to separate out their Hubs? Then another developer comes along and uses the object extensions and can't figure out why their message isn't getting through.

One way to achieve what it looks like you're trying to do is to add the following:

private static readonly _hub = new Hub();
public static Hub Default => _hub;

like you would for a singleton, but without hiding Hubs constructor. Then, users can use the default Hub, but also create a new one if they need it.

Hi,

I successfully upgraded from PubSub 3.1.0 to 4.0.0, but this week I un-installed the app and re-installed from scratch. During startup I received

System.InvalidOperationException has been thrown
Collection was modified; enumeration operation may not execute

Oddly it seemed to crash every time thereafter.

I tried putting in lock mutex's around all of my hub.Subscribe and hub.Publish invocations to ensure no threading issues from my end, but the problem continued.

In the end I had to revert back to PubSub 3.1.0 and then the problem went away.

Unfortunately I don't have the call stack any more, but it was deep inside PubSub with a LINQ .Where clause.

Because your example for subscribing passes this then I recommend changing the unsubscribe documentation to match, i.e. from

public class Page
{
	Hub hub = Hub.Default;
	
	public void WereDoneHere()
	{
		hub.Unsubscribe<Product>();
	}
}

to

public class Page
{
	Hub hub = Hub.Default;
	
	public void WereDoneHere()
	{
		hub.Unsubscribe<Product>(this);
	}
}

Otherwise those using the current example as-is (copy & paste) will likely accidentally unsubscribe all their listeners from the hub in error, rather than from their specific object / subscriber instance.

Hi,

I am using a static pipelinefactory in a dotnet core webapi.
.Exists and Unsubscribe dont seem to be working at all.
Everytime i check and unsubscibe in the constructor but it just keeps adding handlers
Any idea?

        public DashboardHub(
            //PubSubService pubSubService,
            DashboardsService dashboardsService,
            IPubSubPipelineFactory pubSubPipelineFactory,
            IMemoryCache cache)
        {
            //_pubSubService = pubSubService;
            _dashboardsService = dashboardsService;
            _pubSubPipelineFactory = pubSubPipelineFactory;
            _cache = cache;


            if (_pubSubPipelineFactory.GetSubscriber().Exists<OrderDto>(this, OrderDtoChangedHandler))
            {
                // never hits this spot
            }
            

            if (_pubSubPipelineFactory.GetSubscriber().Exists<OrderDto>(this))
            {
                // never hits this spot
            }

            _pubSubPipelineFactory.GetSubscriber().Subscribe<OrderDto>(this, OrderDtoChangedHandler);
            //works ok

        }

Can you add signing so it can be referenced by signed assemblies?

Hi,

Thanks for the library. I do have a pending changes for additional feature to publish message in parallel (this increased throughput for us) and also I added async dispatch (as optional way to broadcast events).

Would you be interested in PR?

I love the initiative of having a lightweight, simple pubsub library. However, I would only use if it saves me from testing the whole publishing feature in my application.

If I had to wrap your lib in a component of my system, I wouldn't be able to mock your Hub component in my unit tests. In other words, I'd need to test the entire feature once again (and duplicate your unit tests ...).

Could you add an IHub to your project and save me the trouble ?
(or make the public methods virtual at least)

Thanks,
Cem

How do we handle the case in which the event handler needs to be async?

Nice little library! Straight to the point.

But I see no reason why it works only from .NET 4.5 onwards. Why not make it work from 3.5 or at least 4.0?

Visual Studio 2012
WinForms .NET 4.5 project .
Tried to install via NuGet and got the error:
'PubSub' could not be installed beacause it is not compatable with any project in the solution. The package doesn't target any framework.

Works fine by deploying the DLL using add reference.

As far as I can see both Publish and PublishAsync use sequential execution. Is there any way to make it run in parallel?

Thanks for the great library!
When using the library NuGet pulls in a ton of other libraries.

Could you also provide a version that depends on .Net Standard 2.0? I think that would alleviate this issue a lot.

Have you thought about targeting NETStandard now?

This is great and I handle the filtering in the subscriber currently, but it would be cool if I had a way to Subscribe to an object, but only if that object has a certain key value. In RabbitMQ I can accomplish it with different channels. Basically I have an object being emitted from a background processor reading from an open socket, completing messages, serializing them as needed, filtering out things that aren't necessary, etc and then eventing/emitting the message out. The message itself is a consistent json payload and I've been experimenting with having that process Publish the Message and then my Consumer objects Subscribe and then each can process that payload however. The only issue is that each of my consumers cares about only payloads that meet certain criteria (generally a single identifier field in the json payload). Currently I just send down all the messages to all the consumers and they filter them as needed, but it would be cool if I could somehow specify in the consumer which payload I want when subscribing. I can't use RabbitMQ or Redis PubSub in this application and was looking for a simple in-memory pubsub mechanism and this has been great.

When we add subscribe code to somewhere like activity onCreate, page load etc, it's possible for this code to be called multiple times. Than, this causes the message action being called multiple times.

I think there should be a precaution or at least an option to prevent this situation.