Detect if a callback has been set for bus.Send

[andreasohlund]

The current api, IBus.Send().Register() doesn't allow us to add headers/etc on the message to tell the server if there is a callback "waiting" for a potential reply since the .Register happens after the send has completed. This info will be needed in the future when we perform directory based routing using ServiceControl to make decisions like `

if IncomingMessage.HasCallbacks then 
    route to exact instance that sent the request
else 
   route to any endpoint instance

I propose that we move ahead with the new bus.Send api suggested in https://github.com/Particular/NServiceBus/issues/1730 and put a callback option there.

Something like this

var options = new SendOptions();

options.OnReply(r=>{...}); 

bus.Send(new MyRequest(),options)

Should we also consider moving the callback support out of the core to make it an explicit opt-in using install-package NServiceBus.Callbacks as described in https://github.com/Particular/NServiceBus/issues/2390 ?

All this could be added in v5.X + old api obsoleted with remove in v6.0

[udidahan]

Given how new v5 is, why not obsolete in v5 and then remove in v6?

[andreasohlund]

My bad, that's what I meant :( Updated

[johnsimons]

If we are considering revamping the callback mechanism we should go back and ask ourselves what the users really want. And I'm pretty sure what they are after is a RPC style of programming over messaging.

With the async cleverness that exists now in the clr, I would suggest us considering instead giving the users what they really want.

At the moment users are doing RPC using the following "technique"/hack:

bus.Send(new MyRequest()).Register(r =>
{
     reply = (MyReply)r.Messages.Single();
});

But what would be really good is something like:

var myReplies = await bus.SendAndWaitForReplies(new MyRequest());
var firstReply = myReplies[0];
...

[danielmarbach]

I'm totally on @johnsimons side. I just wanted to post the same thoughts including:

Think first about an async design and then put a synchronous API on top of it. I'll ask permission to my customer if I can share my WSSB async Bus

[andreasohlund]

Not against async but that's a big change. Can we move forward with the current pipeline?

Can SendAndWaitForReplies be an extension method?

On Wed, Dec 10, 2014 at 8:32 AM, Daniel Marbach notifications@github.com wrote:

I'm totally on @johnsimons https://github.com/johnsimons side. I just wanted to post the same thoughts including:

Think first about an async design and then put a synchronous API on top of it. I'll ask permission to my customer if I can share my WSSB async Bus

— Reply to this email directly or view it on GitHub https://github.com/Particular/NServiceBus/issues/2596#issuecomment-66414139 .

[danielmarbach]

@andreasohlund for whom is it a big change? For NSB internally or for the user?

[danielmarbach]

The problem is the more API you introduce which do not support async, the more you have to back port, obsolete and carry on over multiple versions. I think this cost should also be taken into account

[andreasohlund]

for whom is it a big change? For NSB internally or for the user?

For us, no biggie for the user

[andreasohlund]

The problem is the more API you introduce which do not support async, the more you have to back port, obsolete and carry on over multiple versions. I think this cost should also be taken into account

Agreed

[SzymonPobiega]

SendAndWaitForReplies looks really cool but I am afraid that it does not provide good guidance to the users. If we go with a callback in SendOptions than we make request-reply style a bit more difficult which IMO is a good thing. We don't want our users to build WS-* on top of NServiceBus. Request-reply should only be used when there is no other option.

[johnsimons]

We either accept that our users need this functionality or remove it. Hiding it away in a more obscure api is IMO flaw.

And I'm +1 to remove the current not so helpful enum callback.

On Wednesday, December 10, 2014, Szymon Pobiega notifications@github.com wrote:

SendAndWaitForReplies looks really cool but I am afraid that it does not provide good guidance to the users. If we go with a callback in SendOptions than we make request-reply style a bit more difficult which IMO is a good thing. We don't want our users to build WS-* on top of NServiceBus. Request-reply should only be used when there is no other option.

— Reply to this email directly or view it on GitHub https://github.com/Particular/NServiceBus/issues/2596#issuecomment-66429358 .

[andreasohlund]

Are we all good with splitting the callbacks out to a NServiceBus.Callbacks assembly and making it opt-in? (@udidahan?)

[udidahan]

@andreasohlund you mean in v6?

[andreasohlund]

Yes, v6

7 jan 2015 kl. 18:19 skrev Udi Dahan notifications@github.com:

@andreasohlund you mean in v6?

— Reply to this email directly or view it on GitHub.

[udidahan]

OK then.

[johnsimons]

@andreasohlund how will this affect transports with callback queues ?

[andreasohlund]

how will this affect transports with callback queues ?

In a positive way IMO. Right now they have to have them on to be safe by default. Now they can check if the "callbacks" feature is enabled and only then create the callback queue and start the callback receiver

[distantcam]

After discussing it with @andreasohlund we came up with the following API

var options = new SendOptions();

var replyTask = options.GetReplyTask().ContinueWith(t =>
{
    // Do something with the reply
});

await bus.SendAsync(new Message(), options);

This clearly shows that there's a separation between the send completing, and a reply coming back.

This also allows the user to use the full Task system including checking for faults, continuing on a specific scheduler, etc.

[johnsimons]

What do u plan to expose on the t.result in continuewith ?

[danielmarbach]

From the code flow perspective it would be better to be able to do something like that:

var options = new SendOptions();

var replyTask = options.GetReplyTask();

await bus.SendAsync(new Message(), options);

var response = await replyTask;

//continue here

@andreasohlund and I noted that somewhere in his sublime text two or three weeks ago as a proposal. Thoughts?

From: Cameron MacFarland [mailto:notifications@github.com] Sent: Dienstag, 10. März 2015 12:10 To: Particular/NServiceBus Cc: Daniel Marbach Subject: Re: [NServiceBus] Detect if a callback has been set for bus.Send (#2596)

After discussing it with @andreasohlund https://github.com/andreasohlund we came up with the following API

var options = new SendOptions();

var replyTask = options.GetReplyTask().ContinueWith(t =>

{

// Do something with the reply

});

await bus.SendAsync(new Message(), options);

This clearly shows that there's a separation between the send completing, and a reply coming back.

This also allows the user to use the full Task system including checking for faults, continuing on a specific scheduler, etc.

— Reply to this email directly or view it on GitHub https://github.com/Particular/NServiceBus/issues/2596#issuecomment-78034019 .

[johnsimons]

@danielmarbach what would happened if user does something like:

var options = new SendOptions();

var replyTask = options.GetReplyTask();

await bus.SendAsync(new Message(), options);
await bus.SendAsync(new Message2(), options);
await bus.SendAsync(new Message3(), options);

var response = await replyTask;

//continue here

What is the result ? Would that even work ?

[johnsimons]

I think we need to lock down this API so users cannot stuff it up. I would prefer to have something like:

bus.SendAsync(new Message()).ContinueWith(t=> ...);

I don't think we need a PublishAsync ?

[andreasohlund]

The only 2 must haves for the api is:

1. We need to be able to set a header on the outgoing message indicating that there is a callback set (to enable our routing to adjust for that scenario)
2. It needs to be extendable so that the callbacks can reside in a separate assembly

[danielmarbach]

@johnsimons this can be handled by making the SendOptions GetReplyTask throwing when it is called multiple times or even make it reusable depending where we want to take this.

Another thing which can be achieved (because my proposed solution is more composable):

var options1 = new SendOptions();

var options2 = new SendOptions();

var options3 = new SendOptions();

var replyTask1 = options1.GetReplyTask();

var replyTask2 = options1.GetReplyTask();

var replyTask3 = options1.GetReplyTask();

await bus.SendAsync(new Message(), options1);

await bus.SendAsync(new Message2(), options2);

await bus.SendAsync(new Message3(), options3);

var firstDoneTask = await Task.WhenAny(replyTask1, replyTask2, replyTask3);

var response firstDoneTask.Result;

From: John Simons [mailto:notifications@github.com] Sent: Dienstag, 17. März 2015 01:43 To: Particular/NServiceBus Cc: Daniel Marbach Subject: Re: [NServiceBus] Detect if a callback has been set for bus.Send (#2596)

@danielmarbach https://github.com/danielmarbach what would happened if user does something like:

var options = new SendOptions();

var replyTask = options.GetReplyTask();

await bus.SendAsync(new Message(), options);

await bus.SendAsync(new Message2(), options);

await bus.SendAsync(new Message3(), options);

var response = await replyTask;

//continue here

What is the result ? Would that even work ?

— Reply to this email directly or view it on GitHub https://github.com/Particular/NServiceBus/issues/2596#issuecomment-82006418 .

[johnsimons]

At the moment all proposed api do not mention what the return value is, we currently support return an enum (int) and also a convoluted was to get to the messages array I would like to propose as part of this change to remove the return enum and instead only support returning messages. This would mean:

• Users get what they have always wanted in a more fluid api, RPC style calls;
• We can get rid of bus.Return, has that is not needed;

Thoughts?

[johnsimons]

We need to be able to set a header on the outgoing message indicating that there is a callback set (to enable our routing to adjust for that scenario)

@andreasohlund is this another YAGNI ?

[distantcam]

Ok, what about this?

var responseHandle = await bus.SendAsync(new Message()); // Returns immediately

responseHandle.ReplyTask.ContinueWith(/* continuation */);

The reply task only completes when a reply is received. This solves the problem of multiple sends with the same option object and is a nicer API for the user.

Only downside is that there's a subtle implication that the send has completed because it has returned, when that's not actually true.

@johnsimons I had the return value being the value coming from the reply task. I believe that's how it currently works (the callback is given the result).

[johnsimons]

On a call with @distantcam we had another go at designing the api, here is what we came up with:

public async Task<IHttpActionResult> GetOrders()
{
    SendContext response = bus.SendWithReply(new Message());

    //response.ReplyTask.ContinueWith()...
    CompletionResult result = await response.ReplyTask;

    MyMessage msg = result.Messages[0]

    return Ok(msg);
}

[danielmarbach]

So SendAsync is gone?

Am Donnerstag, 19. März 2015 schrieb John Simons :

On a call with @distantcam https://github.com/distantcam we had another go at designing the api, here is what we came up with:

public async Task GetOrders() { SendContext response = bus.Send(new Message());

//response.ReplyTask.ContinueWith()...
CompletionResult result = await response.ReplyTask;

MyMessage msg = result.Messages[0]

return Ok(msg);

}

— Reply to this email directly or view it on GitHub https://github.com/Particular/NServiceBus/issues/2596#issuecomment-83352927 .

Daniel Marbach

Solution Architect, Particular Software

Email: daniel.marbach@particular.net

Web: www.particular.net http://t.signauxneuf.com/e1t/c/5/f18dQhb0S7lC8dDMPbW2n0x6l2B9nMJN7t5XX4RzDjlVS9QB88pTc_dW3Ljg7x56dWyBdgBXs402?t=http%3A%2F%2Fwww.particular.net%2F&si=4896787407044608&pi=33cc6202-1f31-4040-8d6c-e70641f44bba

[johnsimons]

Yes @danielmarbach We think using a name like SendAsync could give the wrong impression (the sending is being performed in an async way which is not true).

[danielmarbach]

Why don't we add different methods?

Like Task SendAsync and Task SendWithResponseAsync()

It would be cool to have a oneliner :)

Am Donnerstag, 19. März 2015 schrieb John Simons :

On a call with @distantcam https://github.com/distantcam we had another go at designing the api, here is what we came up with:

public async Task GetOrders() { SendContext response = bus.Send(new Message());

//response.ReplyTask.ContinueWith()...
CompletionResult result = await response.ReplyTask;

MyMessage msg = result.Messages[0]

return Ok(msg);

}

— Reply to this email directly or view it on GitHub https://github.com/Particular/NServiceBus/issues/2596#issuecomment-83352927 .

Daniel Marbach

Solution Architect, Particular Software

Email: daniel.marbach@particular.net

Web: www.particular.net http://t.signauxneuf.com/e1t/c/5/f18dQhb0S7lC8dDMPbW2n0x6l2B9nMJN7t5XX4RzDjlVS9QB88pTc_dW3Ljg7x56dWyBdgBXs402?t=http%3A%2F%2Fwww.particular.net%2F&si=4896787407044608&pi=33cc6202-1f31-4040-8d6c-e70641f44bba

[danielmarbach]

What about the future when we support async?

Am Donnerstag, 19. März 2015 schrieb Daniel Marbach :

Why don't we add different methods?

Like Task SendAsync and Task SendWithResponseAsync()

It would be cool to have a oneliner :)

Am Donnerstag, 19. März 2015 schrieb John Simons :

On a call with @distantcam https://github.com/distantcam we had another go at designing the api, here is what we came up with:

public async Task GetOrders() { SendContext response = bus.Send(new Message());

//response.ReplyTask.ContinueWith()...
CompletionResult result = await response.ReplyTask;

MyMessage msg = result.Messages[0]

return Ok(msg);

}

— Reply to this email directly or view it on GitHub https://github.com/Particular/NServiceBus/issues/2596#issuecomment-83352927 .

Daniel Marbach

Solution Architect, Particular Software

Email: daniel.marbach@particular.net javascript:_e(%7B%7D,'cvml','daniel.marbach@particular.net');

Web: www.particular.net http://t.signauxneuf.com/e1t/c/5/f18dQhb0S7lC8dDMPbW2n0x6l2B9nMJN7t5XX4RzDjlVS9QB88pTc_dW3Ljg7x56dWyBdgBXs402?t=http%3A%2F%2Fwww.particular.net%2F&si=4896787407044608&pi=33cc6202-1f31-4040-8d6c-e70641f44bba

Daniel Marbach

Solution Architect, Particular Software

Email: daniel.marbach@particular.net

Web: www.particular.net http://t.signauxneuf.com/e1t/c/5/f18dQhb0S7lC8dDMPbW2n0x6l2B9nMJN7t5XX4RzDjlVS9QB88pTc_dW3Ljg7x56dWyBdgBXs402?t=http%3A%2F%2Fwww.particular.net%2F&si=4896787407044608&pi=33cc6202-1f31-4040-8d6c-e70641f44bba

[distantcam]

@danielmarbach The SendWithResponseAsync(), what's being awaited here? Is it the send or the reply? If an error occurs was it caused by the send, or the reply?

Given there are 2 different concepts here an API that distinguishes between them is better than an API that shoves everything into a single line because that would be cooler.

[danielmarbach]

Good point cameron! Because I'm not yet convinced about the proposals I'm brainstorming

Am Donnerstag, 19. März 2015 schrieb Cameron MacFarland :

@danielmarbach https://github.com/danielmarbach The SendWithResponseAsync(), what's being awaited here? Is it the send or the reply? If an error occurs was it caused by the send, or the reply?

Given there are 2 different concepts here an API that distinguishes between them is better than an API that shoves everything into a single line because that would be cooler.

— Reply to this email directly or view it on GitHub https://github.com/Particular/NServiceBus/issues/2596#issuecomment-83368772 .

Daniel Marbach

Solution Architect, Particular Software

Email: daniel.marbach@particular.net

Web: www.particular.net http://t.signauxneuf.com/e1t/c/5/f18dQhb0S7lC8dDMPbW2n0x6l2B9nMJN7t5XX4RzDjlVS9QB88pTc_dW3Ljg7x56dWyBdgBXs402?t=http%3A%2F%2Fwww.particular.net%2F&si=4896787407044608&pi=33cc6202-1f31-4040-8d6c-e70641f44bba

[danielmarbach]

@distantcam I thought about it again. I changed my mind. We are a library/abstraction layer on top of the queues. Why do we need to bother users with Continuations in order to get the response. I think we can handle the continuation inside the library and give them back a task which can await the response. This reduces nesting of continuations and makes the code more readable which is a benefit:

interface IBus
{
    Task SendAsync(object message, SendContext context);

    Task<TResponse> SendAndAwaitResponseAsync<ResponseMessage>(object message, SendContext context);
}

public class SomeController : ApiController 
{
    public Task<IActionResult> Get() {
        var response = await bus.SendAndAwaitResponseAsync(message, new SendContext());
    }
}

I know we don't want to extend IBus. I just used it as an example

[andreasohlund]

is this another YAGNI ?

No, when me and @udidahan was going through how routing would work when there are callbacks set we came to the conclusion that the routing needs to know if there are callbacks to make sure responses are routed to the correct instance. Happy to talk that through on a call if needed!

[johnsimons]

@danielmarbach I think we are pretty much aligned, the example you have provided is pretty much the same as what Cam and I came up with

[johnsimons]

@andreasohlund are you talking about the current routing implementation or a future version ?

[distantcam]

@johnsimons No the sample provided by @danielmarbach is not the same as mine.

Using @danielmarbach's example

public class SomeController : ApiController 
{
    public Task<IActionResult> Get() {
        try {
            var response = await bus.SendAndAwaitResponseAsync(message, new SendContext());
            return Ok(response);
        } catch (Exception ex) {
            // Is this exception because of the send, or the reply?
        }
    }
}

The subtle distinction is that in my example the reply task is separate from the send method, as this is an important difference for the user.

Here's my take.

public class SomeController : ApiController 
{
    public Task<IActionResult> Get() {
        var sendResult = await bus.SendWithContextAsync(message, new SendContext())
            .ReportError("An error occurred during sending");
        var result = await sendResult.ReplyTask
            .ReportError("An error occurred during a response");
        return Ok(result);
    }
}

public static class TaskExtensions
{
    public static Task<T> ReportError<T>(this Task<T> task, string errorMessage)
    {
        var tcs = new TaskCompletionSource<T> ();
        task.ContinueWith (ant =>
        {
            if (ant.IsFaulted)
            {
                Console.WriteLine(errorMessage + ant.Exception.InnerException);
                tcs.SetException(ant.Exception.InnerException);
            }
            else if (ant.IsCanceled)
                tcs.SetCanceled();
            else
                tcs.SetResult(ant.Result);
        });
        return tcs.Task;
    }
}

[andreasohlund]

@johnsimons the one we're building right now

[johnsimons]

Done as part of #2689