Issues Publishing From Subsequent Rebus Instances

Question

I have several services that are essentially console applications hosted using TopShelf, and communiate using Rebus 0.99.50. One of these services (StepManager) loops through a collection of objects (of type Step), each of which contains a Bus instance, which it uses to send a message, and a handler used to handle a reply. The following Step(s) used for this example, in this order, are:

• ReceiveFile
• LogFileMetrics
• ArchiveIncomingFile

In my actual scenario, I have a total of 7 Step(s)...When looping through these Step(s), ReceiveFile and LogFileMetrics behave as expected, however when ArchiveIncomingFile runs, .Send(req) is called, but the message never reaches its destination, leaving the process waiting for the reply that never returns. Regardless of what type of Step object or order of the objects in the list, this happens consistently at second instance of type Step (which does a .Send(req) in the Run() method) in the list. BUT, when I comment out the while (!Completed) { await Task.Delay(25); } statements, the messages appear to get sent, however without those statements, the Step(s) will all run with no specific execution order, which is a problem.

Why is this happening? What am I missing/doing wrong here? And is there a better alternative to accomplish what I am trying to do?

Here are the relevant portions of the classes in question:

public class StepManager
{
    ...

    public string ProcessName { get; set; }
    public List<Step> Steps { get; set; }
    public BuiltinHandlerActivator ServiceBus { get; set; }

    ... 

    public async Task Init()
    {
        ...

        Steps = new List<Step>();
        var process = Db.Processes.Include("Steps")
                                .Where(p => p.Name == ProcessName)
                                .FirstOrDefault();
        ...

        foreach (var s in process.Steps)
        {
            var step = container.Resolve<Step>(s.Name);

            ...

            Steps.Add(step);
        }      
    }

    public async Task Run()
    {
        foreach (var step in Steps)
        {
            await step.Run();
        }
    }
}

public class Step 
{
    public BuiltinHandlerActivator ServiceBus { get; set; }

    public Step()
    {
        Db = new ClearStoneConfigContext();
        Timer = new Stopwatch();
        StepId = Guid.NewGuid().ToString();

        Completed = false;
    }

    public virtual async Task Run() { }
}

public class ReceiveFile : Step
{
    public ReceiveFile()
    {
        ServiceBus = new BuiltinHandlerActivator();

        Configure.With(ServiceBus)
                .Logging(l => l.ColoredConsole(LogLevel.Info))                    
                .Routing(r => r.TypeBased().Map<ProcessLog>("stepmanager"))                    
                .Transport(t => t.UseMsmq("receivefile"))
                .Start();
    }

    public override async Task Run()
    {
        ...

        LogEntry.Message = "File " + FileEvent.Name + " received.";     
        await ServiceBus.Bus.Advanced.Routing.Send("stepmanager", LogEntry);
        Completed = true;           
    }
}

public class LogFileMetrics : Step
{
    public LogFileMetrics()
    {
        SubscriptionTable = "SandboxServiceBusSubscriptions";
        ServiceBus = new BuiltinHandlerActivator();

        Configure.With(ServiceBus)
                .Logging(l => l.ColoredConsole(LogLevel.Info))                    
                .Routing(r => r.TypeBased().Map<LogFileMetricsRequest>("metrics"))                   
                .Transport(t => t.UseMsmq("logfilemetrics"))
                .Start();

        ServiceBus.Handle<FileMetricsLogged>(async msg=> await FileMetricsLogged(msg));;
    }

    public override async Task Run()
    {
        ...

        await ServiceBus.Bus.Send(new LogFileMetricsRequest { ProcessId = ProcessId, FileEvent = FileEvent }).ConfigureAwait(false);

        while (!Completed) { await Task.Delay(25); }        
    }

    private async Task FileMetricsLogged(FileMetricsLogged msg)
    {
        ...

        await ServiceBus.Bus.Advanced.Routing.Send("stepmanager", LogEntry);
        Completed = true;
    }
}

public class ArchiveIncomingFile : Step
{
    public ArchiveIncomingFile()
    {
        SubscriptionTable = "SandboxServiceBusSubscriptions";
        ServiceBus = new BuiltinHandlerActivator();

        Configure.With(ServiceBus)
                .Logging(l => l.ColoredConsole(LogLevel.Info))                    
                .Routing(r => r.TypeBased().Map<ArchiveIncomingFileRequest>("incomingarchivefilerouter"))                    
                .Transport(t => t.UseMsmq("archiveincomingfile"))                    
                .Start();

        ServiceBus.Handle<IncomingFileArchived>(async msg => await IncomingFileArchived(msg));
    }

    public override async Task Run()
    {
        ...

        ServiceBus.Bus.Send(req);

        while (!Completed) { await Task.Delay(25); }
    }

    private async Task IncomingFileArchived(IncomingFileArchived msg)
    {
        ...

        await ServiceBus.Bus.Advanced.Routing.Send("stepmanager", LogEntry);
        Completed = true;
    }
}

Answer 1

I can see several issues with your code, although it is not clear to me what is causing the funny behavior you are experiencing.

First off, it seems like you are creating new bus instances every time you are creating steps. Are you aware that Rebus' bus instance is supposed to be created once at startup in your application, kept as a singleton, and must be properly disposed when your application shuts down?

You can of course perform this create-dispose cycle as many times as you like, it's not like Rebus will leave anything behind in any way, but the fact that you are NOT disposing the bus anywhere tells me that your application probably forgets to do this.

You can read more on the Rebus wiki, especially in the section about Rebus' bus instance.

Another issue is the subtle potential race condition in the ArchiveIncomingFile class whose ctor looks like this:

public ArchiveIncomingFile()
{
    SubscriptionTable = "SandboxServiceBusSubscriptions";
    ServiceBus = new BuiltinHandlerActivator();

    Configure.With(ServiceBus)
            .Logging(l => l.ColoredConsole(LogLevel.Info))                    
            .Routing(r => r.TypeBased().Map<ArchiveIncomingFileRequest>("incomingarchivefilerouter"))                    
            .Transport(t => t.UseMsmq("archiveincomingfile"))                    
            .Start();

    //<<< bus is receiving messages at this point, but there's no handler!!

    ServiceBus.Handle<IncomingFileArchived>(async msg => await IncomingFileArchived(msg));
}

As you can see, there is a (very very very short, admittedly) time (marked by //<<<) in which the bus has been started (and thus will start to pull messages out of its input queue) where no handlers yet have been configured.

You should be sure to configure handlers BEFORE you start the bus.

Finally, you are asking

And is there a better alternative to accomplish what I am trying to do?

but I am unable to answer that question because I simply cannot figure out what you are trying to do ;)

(but if you explain to me at a slightly higher level what problem you are trying to solve, I might have some hints for you :))