Missed Broadcast Events

[akytsun]

Describe the bug In my net core 2.2 web api project I'm using _queue.QueueBroadcast to publish events. I've added logging to my IListener<MyEvent> implementation in HandleAsync method. ConsummationDelay was set to 15. Every time if I trigger api call which trigger _queue.QueueBroadcast(new MyEvent()) one by one, only one of published events will be really consumed and fired. Where Coravel can miss others I don't know. Any ideas?

Affected Coravel Feature Queuing, Event Broadcasting

Expected behaviour HandleAsync should fire for each published events at max ConsummationDelay time

[jamesmh]

Nope... I'll take a look at this in next few days and get back to you. What if you remove the first listener - does it still only fire one?

[jamesmh]

Perhaps you didn't register all the listeners with the DI container?

[akytsun]

at Configure Services I've added next:

services.AddQueue();
services.AddEvents();
services.AddTransient<MyListener>();

at Configure:

 var provider = app.ApplicationServices;
            provider.ConfigureQueue().LogQueuedTaskProgress(provider.GetService<ILogger<IQueue>>()).OnError(ex =>
            {
                provider.GetService<ILogger<IQueue>>().LogError(ex, "Smth wrong in Coravel");
            });

            IEventRegistration registration = provider.ConfigureEvents();

            registration
            .Register<MyEvent>()
            .Subscribe<MyListener>();

I have only one Listener registered(

I've tested it multiple times and I assume that after ConsummationDelay _queue is clearing after that Or I can wrong

[akytsun]

Also, I can reproduce a similar behavior but use QueueInvocable. Not all registered calls were invoked

[jamesmh]

Yes, after the consummation delay the entire queue should clear. I'll see if I can get a sample running to verify (this should be working though...)

[jamesmh]

I created a simple sample to test these things:

https://github.com/jamesmh/coravel-multi-queue-sample

[jamesmh]

If that still doesn't help I think you'll have to post your listener code

[akytsun]

James, I'm not using scheduler for publish events. I'm using queue.BroadcastEvent in my apicontroller action method. Maybe problem with queue instance which injected through DI into controller?

Пн, 20 січ. 2020, 17:59 користувач James Hickey notifications@github.com пише:

If that still doesn't help I think you'll have to post your listener code

— You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub https://github.com/jamesmh/coravel/issues/132?email_source=notifications&email_token=AOADSMCG5QJTKBQVZ2K4L33Q6XC4TA5CNFSM4KIUV4K2YY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEJNC27I#issuecomment-576335229, or unsubscribe https://github.com/notifications/unsubscribe-auth/AOADSME7YZTEZJR23POZVQLQ6XC4TANCNFSM4KIUV4KQ .

[jamesmh]

Yes, I know. The scheduler is just to simulate someone queuing via the system. That same code could be moved to a controller and should work.

You can comment out the scheduler code in Startup.cs and paste this into the sample controller:

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Logging;
using coravelMultiQueueSample.Models;
using coravelMultiQueueSample.Events;
using coravelMultiQueueSample.Invocables;
using Coravel.Queuing.Interfaces;

namespace coravelMultiQueueSample.Controllers
{
    public class HomeController : Controller
    {
        private readonly ILogger<HomeController> _logger;
        private readonly IQueue _queue;

        public HomeController(ILogger<HomeController> logger, IQueue queue)
        {
            _logger = logger;
            _queue = queue;
        }

        public IActionResult Index()
        {
            _queue.QueueBroadcast(new TestEvent());
            _queue.QueueInvocable<InvocableOne>();
            _queue.QueueInvocable<InvocableTwo>();
            _queue.QueueInvocable<InvocableThree>();
            return View();
        }

        public IActionResult Privacy()
        {
            return View();
        }

        [ResponseCache(Duration = 0, Location = ResponseCacheLocation.None, NoStore = true)]
        public IActionResult Error()
        {
            return View(new ErrorViewModel { RequestId = Activity.Current?.Id ?? HttpContext.TraceIdentifier });
        }
    }
}

You'll see that hitting the index page will trigger those queued items and they will all fire within the 30 second consummation delay.

[jamesmh]

Did you still need help with this?