Masstransit: Integrate external DI scope for ISendEndpointProvider.

[Crozin]

MT provides great integration for DI containers when run as receive endpoint. Custom filters added to consume pipeline can make use of services accessible form IServiceProivder. They're also ran in within scope of DI container. Filters added to send/publish pipelines can also make good use of those if they're triggered from consume pipeline as they have access to wrapping ConsumeContext and its IServiceProvider. However when using a send-only bus, where we publish something from DI top-level scope (e.g. ASP.NET HostedService) or from non-MT initiated DI scope (e.g. ASP.NET HTTP Controller) using IBus send/publish filters can't access ConsumeContext as such does not exist.

I'm wondering if/how we could bring DI-scoped IServiceProvider to send filters in such a case? Consider following ASP.NET Controller:

public class MyController : ApiController
{
  private readonly IBus bus;

  // constructor

  public async Task<string> DoSth()
  {
    await bus.Publish(...); // should behave as current implemention
    await bus.AbcScoped().Publish(...); // could run IServiceProivder.CreateScope() under the hood
    await bus.AbcExternallyScoped(RequestServices).Publish(...); // could run with externally provided `IServiceProvider`

    return "OK";
  }
}

This would require some "proxy ISendEndpointProivder/IPublishEndpointProvider" to be returned by IBus.Abc...Scoped() which would set up DI scope and before sending anything through send/publish pipelines I guess?

[phatboyg]

I think this could/should be done in a way that IBus, ISendEndpointProvider, and IPublishEndpoint are decorated coming out of the container as Scoped when in a controller/Request Scope. It's a different scope, it's a controller, but it's still a scope. There just isn't a consume context available.

[NooNameR]

What is the real use case for this?

[phatboyg]

Consumer dependencies want to publish events or send messages, requests, etc. need to get a scoped version of ISendEndpointProvider or IPublishEndpoint so that message headers are properly set on the outgoing messages. Using IBus inside of a ConsumeContext is pure evil and should be avoided at all costs.

[NooNameR]

basically that's why I am asking, just do not use IBus/IBusControl :)

[phatboyg]

You should only use IBus when you aren't in a consumer, saga, or a dependency.
You should only use IBusControl if you're the program starting/stopping the bus.

[phatboyg]

Could you just use the IServiceProvider, call CreateScope() and manage it that way to resolve your IPublishEndpoint or ISendEndpointProvider? There would not be a ConsumeContext but it would be in a scope.

[Crozin]

OK, so I'm back after some Easter break.

Initially I though it gonna be quite similar to transaction outbox in implementation, but turned out to be little more difficult to implement. For sake of simplicity let's assume quite obvious case where we have ASP.NET Core API Controller action that sends a message on a queue.

class AbcController : Controller
{
    public async Task<string> DoSth()
    {
        var scope = HttpContext.RequestServices; // scoped IServiceProvider 
        var sep = new ScoperSendEndpointProivder(scope, scope.GetRequiredService<ISendEndpointProvider>());
        var se = await sep.GetSendEndpoint(...);

        await se.Send(...);

        return "OK";
    }
}

class ScoperSendEndpointProivder : ISendEndpointProvider
{
    // constructor & some interface methods omitted

    public Task<ISendEndpoint> GetSendEndpoint(Uri address)
    {
        return new ScopedSendEndpoint(scope, await sep.GetSendEndpoint(address));
    }
}

class ScopedSendEndpoint : ISendEndpoint
{
    private readonly ISendEndpoint sendEndpoint;
    private readonly IServiceProvider scope;

    // constructor & some interface methods omitted

    public Task Send<T>(T message, IPipe<SendContext<T>> pipe, CancellationToken cancellationToken = default) where T : class
    {
        return sendEndpoint.Send(message, new ScopedSendPipe<T>(scope, pipe), cancellationToken);
    }
}

class ScopedSendPipe<T> : IPipe<SendContext<T>> where T : class
{
    public async Task Send(SendContext<T> context)
    {
        context.GetOrAddPayload(() => scope);

        await pipe.Send(context);
    }
}

Ufff... that's quite a hierarchy to pass objects through, but final context.GetOrAddPayload(() => scope); should do the trick, right? Nope. Well... almost. It will definitely make scoped service provider available further down the pipeline. However it has no effect on any filters registered for send/publish pipes on transport level as those are executed prior to any "per-message" pipes.

One thing I'm thinking of is creation of send/publish filter that will use HttpContextAccessor and will check if it's called from non-ConsumeContext environment. This has obvious disadvantage. Gonna work only for HTTP-based scopes. In case of different scope environment I'd have to crate a copy of HttpContext and HttpContextAccessor.

Please correct me if I'm wrong. Is it even a step towards good direction?

[phatboyg]

There is a special interface, ISendContextPipe which is called prior to the middleware pipelines of Send, which is designed for things like this. It's how the ConsumeContext is added.

If you add that as an additional interface on your ScopedSendPipe, you can add the payload.

[NooNameR]

@Crozin you are trying to achieve Scope container inside your filters or what? could you give me please more information what is your goal

[phatboyg]

Yes. He wants to be able to get the container scope from a send/publish filter when not consuming a message.

[NooNameR]

what I did before, you have control from where you are sending a message, so I've been using IHttpContextAccessor inside my send filters, and you can easily check if _accessor.HttpContext == null it is fair enough

[NooNameR]

also you can create send filter using ISendPipeSpecificationObserver with IFilter<SendContext<T>> which will create scope and put it into payload using root provider or httpcontext if available, and use this filter as first in your list

[phatboyg]

also you can create send filter using ISendPipeSpecificationObserver

That's literally what he did above. The issue is you can't have it global, because the scope doesn't exist yet. It needs to decorate the send endpoint like the transaction out box does. I think he's close, it's the ISendContextPipe that's needed to execute it first.

[NooNameR]

public class ScopeSendPipeSpecificationObserver :
        ISendPipeSpecificationObserver
    {
        private readonly IServiceProvider _serviceProvider;

        public ScopeSendPipeSpecificationObserver(IServiceProvider serviceProvider)
        {
            _serviceProvider = serviceProvider;
        }

        public void MessageSpecificationCreated<T>(IMessageSendPipeSpecification<T> specification) where T : class
        {
            specification.UseFilter(new ScopeSendFilter<T>(_serviceProvider));
        }
    }

    public class ScopeSendFilter<T> : IFilter<SendContext<T>> where T : class
    {
        private readonly IServiceProvider _serviceProvider;

        public ScopeSendFilter(IServiceProvider serviceProvider)
        {
            _serviceProvider = serviceProvider;
        }

        public async Task Send(SendContext<T> context, IPipe<SendContext<T>> next)
        {
            var httpContextAccessor = _serviceProvider.GetService<HttpContextAccessor>();
            var serviceProvider = httpContextAccessor.HttpContext?.RequestServices ?? _serviceProvider;
            context.GetOrAddPayload(() => serviceProvider.CreateScope());
        }

        public void Probe(ProbeContext context)
        {
        }
    }

for bus configuration use as a first filter:

cfg.ConfigureSend(cfg => cfg.ConnectSendPipeSpecificationObserver(new ScopeSendPipeSpecificationObserver(provider)));

[NooNameR]

@phatboyg
his ScopedSendPipe always will be as a latest pipe

[phatboyg]

That's cool - but isn't there already a request scope?

It would be a nice AspNetCore addition to register something that adds this to the send/publish automatically for controllers that resolve those types.

[NooNameR]

That's cool - but isn't there already a request scope?

did not get

It would be a nice AspNetCore addition to register something that adds this to the send/publish automatically for controllers that resolve those types.

I could do it if you want

[phatboyg]

A controller is resolved within an HTTP request scope, so the scope is already created. That scope/service provider should be added to the pipe, not create a new one. That way the filters in the send/publish pipes are using the same scope as the HTTP controller.

[NooNameR]

could be, the only issue how we can configure Send if we do not have control on bus configuration? do you have any ideas?

[phatboyg]

I only care about things done within .AddMassTransit() - as far as being able to configure the registered components, etc. We'd have to give them a method to configure it. Something like:

cfg.UseHttpRequestScope()

To configure the send/publish provider injection to the payloads.

[NooNameR]

Have created quick draft, what do you think @phatboyg
https://github.com/MassTransit/MassTransit/pull/1815

[Crozin]

There is a special interface, ISendContextPipe which is called prior to the middleware pipelines of Send, which is designed for things like this. It's how the ConsumeContext is added.

Yeah, that was it! :)

---

I might actually share some I'm working on right now:

    public interface IScopedSendEndpointProvider : ISendEndpointProvider { }

    public class ScopedSendEndpointProvider : IScopedSendEndpointProvider
    {
        private readonly ISendEndpointProvider sendEndpointProvider;
        private readonly IServiceProvider scope;

        public ScopedSendEndpointProvider(ISendEndpointProvider sendEndpointProvider, IServiceProvider scope)
        {
            this.sendEndpointProvider = sendEndpointProvider;
            this.scope = scope;
        }

        public ConnectHandle ConnectSendObserver(ISendObserver observer) => throw new NotImplementedException();

        public async Task<ISendEndpoint> GetSendEndpoint(Uri address)
        {
            return new ScopedSendEndpoint(await sendEndpointProvider.GetSendEndpoint(address), scope);
        }
    }

    public class ScopedSendEndpoint : ISendEndpoint
    {
        private readonly ISendEndpoint sendEndpoint;
        private readonly IServiceProvider scope;

        public ScopedSendEndpoint(ISendEndpoint sendEndpoint, IServiceProvider scope)
        {
            this.sendEndpoint = sendEndpoint;
            this.scope = scope;
        }

        public ConnectHandle ConnectSendObserver(ISendObserver observer) => throw new NotImplementedException();

        public Task Send<T>(T message, CancellationToken cancellationToken = new CancellationToken()) where T : class
        {
            return sendEndpoint.Send(message, new ScopedSendPipe<T>(scope), cancellationToken);
        }

        public Task Send<T>(T message, IPipe<SendContext<T>> pipe, CancellationToken cancellationToken = default) where T : class
        {
            return sendEndpoint.Send(message, new ScopedSendPipe<T>(scope, pipe), cancellationToken);
        }

        public Task Send<T>(T message, IPipe<SendContext> pipe, CancellationToken cancellationToken = default) where T : class
        {
            return sendEndpoint.Send(message, new ScopedSendPipe<T>(scope, pipe), cancellationToken);
        }

        public Task Send(object message, CancellationToken cancellationToken = new CancellationToken()) => throw new NotImplementedException();

        public Task Send(object message, Type messageType, CancellationToken cancellationToken = new CancellationToken()) => throw new NotImplementedException();

        public Task Send(object message, IPipe<SendContext> pipe, CancellationToken cancellationToken = new CancellationToken()) => throw new NotImplementedException();

        public Task Send(object message, Type messageType, IPipe<SendContext> pipe, CancellationToken cancellationToken = new CancellationToken()) => throw new NotImplementedException();

        public Task Send<T>(object values, CancellationToken cancellationToken = new CancellationToken()) where T : class => throw new NotImplementedException();

        public Task Send<T>(object values, IPipe<SendContext<T>> pipe, CancellationToken cancellationToken = new CancellationToken()) where T : class => throw new NotImplementedException();

        public Task Send<T>(object values, IPipe<SendContext> pipe, CancellationToken cancellationToken = new CancellationToken()) where T : class => throw new NotImplementedException();
    }

    internal class ScopedSendPipe<TMessage> : IPipe<SendContext<TMessage>>, ISendContextPipe where TMessage : class
    {
        private readonly IServiceProvider scope;
        private readonly IPipe<SendContext<TMessage>> pipe;

        public ScopedSendPipe(IServiceProvider scope, IPipe<SendContext<TMessage>> pipe = null)
        {
            this.scope = scope;
            this.pipe = pipe;
        }

        public Task Send(SendContext<TMessage> context)
        {
            return global::GreenPipes.PipeExtensions.IsNotEmpty(pipe) ? pipe.Send(context) : Task.CompletedTask;
        }

        public void Probe(ProbeContext context)
        {
            pipe?.Probe(context);
        }

        public Task Send<T>(SendContext<T> context) where T : class
        {
            context.GetOrAddPayload(() => scope);

            return Task.CompletedTask;
        }
    }

And in some startup code:

services.AddScoped<IScopedSendEndpointProvider, ScopedSendEndpointProvider>()

And finally from some scoped service like HTTP controller:

class MyClass
{
  private readonly IScopedSendEndpointProvider scopedSendEndpointProvider;

  public MyClass(IScopedSendEndpointProvider scopedSendEndpointProvider)
  {
    this.scopedSendEndpointProvider = scopedSendEndpointProvider;
  }

  public async Task MyAction()
  {
    await sendEndpoint = await scopedSendEndpointProvider.GetSendEndpoint(...);
    await sendEndpoint.Send(...);
  }
}

Key benefit over approach where HttpContext/HttpContextAccessor is used is fact that we're no longer bound to this very specific "container scope" of ASP.NET HTTP request but we can work with any generic container scope. That's in fact my case we I'm not dealing with HTTP at all but need such a scoped functionality.

[phatboyg]

And publish endpoint can be done the same way. This _feels_ a lot cleaner, less code, easy to understand, and matches how it's done by the consume pipeline.

[NooNameR]

@phatboyg so we should discard this one?
https://github.com/MassTransit/MassTransit/pull/1815

EDIT:
basically, for PR we have more control for scope and what we can inject, and we could use HttpContext scope, etc

[phatboyg]

So the PR is merged, @Crozin does it work for what you need (the PR changes).

[Crozin]

@phatboyg @NooNameR As far as I see this PR handles HttpContext and "internal" scopes only whereas what I need is a solution for generic "external" container scopes.

[NooNameR]

@Crozin MT creates it own scope for every publish/send, the only case when you are using HttpContext it will be used

[Crozin]

@NooNameR Yeah, but I'm not dealing with HTTP environment so HttpContext is useless in my case. I'm also not interested in creating brand new scope as I already have one. This is why what I really wanted is ability to pass externally owned DI scope and I've managed to achieve that using solution I've posted above.

This PR will suit tone of usecases but won't help me. :)

[NooNameR]

what are you using to create your scope?

[Crozin]

what are you using to create your scope?

Well the container itself obviously. :) Basically I'm working on network app that's not a HTTP/web application. Therefore whenever a request/event is received what I do is actually something quite similar to ASP.NET Core HTTP pipeline handling. And there are like three completely separate pipelines like that. Each and every one of them would require its own HttpContext-like object and HttpContextAccessor-like service.

As I said I think this PR was useful as HTTP environment is fairly common but it simply does not suit my criteria at all. :)

[NooNameR]

do you have anything dependant on a current scope?
basically you could use asyncLocal variable to make something similar to IHttpContextAccessor but yours, and use AspNetCore sample to create your own ISendScopeProvider and IPublishScopeProvider

[phatboyg]

@Crozin So the code snipped above that you were using to add your own scope using ISendContextPipe, that is working for you? That approach anyway? It should be fairly easy to allow an existing scope to be specified on the front-end to SendEndpoint/PublishEndpoint.

[Crozin]

do you have anything dependant on a current scope?

Yeah, some scope-bound services, hence the scope in the first place. I'm aware I could mimic IHttpContextAccessor and create custom classes like and with corresponding filters like those you've added for HttpContext, but I'd rather use a generic approach which clearly says "it's about DI scope, not some XyzAccessor to workaround".

@phatboyg Yup, it works quite well. Example of MyClass from code snipped I've posted uses DI container but it could be done manually as well.

IServiceProvider sp; // root or nested scope
ISendEndpointProvider sep; // MassTransitBus or whatever

using var scope = sp.CreateScope();
var scopedSep = new ScopedSendEndpointProvider(sep, scope.ServiceProvider);
var se = await scopedSep.GetSendEndpoint(...);

await se.Send(...);

There is one gotcha. Use of in-memory outbox is tricky as it is usually placed above container creation in pipeline. This obviously means no send/publish filter can use DI scope as it is already disposed at the time of outbox flush. I guess I'd have to move outbox below container management in pipeline, however I just realised I can simply get rid of that as my system effectively doesn't need it so I haven't dig into that yet.

BTW this might be some offtopic but we have ISendContextPipe but no IPublishContextPipe?

[NooNameR]

@Crozin there is multiple issues with a code you posted, in your case you are not implementing multiple methods, also, you do not have ability to create scope if it is not scoped call, it is more scopish

you can your own scope provider which will be using AsyncLocal<IServiceScope> to set value, and create sort of middleware which on start of your request create new scope and put it inside asyncLocal variable, after that your scope provide will read value from there and put it inside payload, if you want, I can make a sample for you

[Crozin]

@NooNameR Obviously code I've posted above is incomplete as it served only as a reference/example. I didn't bother with some extra logic to create a scope as it was not something I've originally desired.

I see no reason to force this AsyncLocal<IServiceScope> idea. It creates global/hidden dependency for the code. Does not scale well (think multiple nested scopes). I'd really prefer to explicitly pass scope object just as any other data in my pipeline.

[NooNameR]

tbh, I'm not sure If I could help you, the idea behind MT: it creates it own scope everywhere.
it seems your case is very specific :)

[phatboyg]

It's actually very specific. I'll handle it.

[NooNameR]

should be solved by: https://github.com/MassTransit/MassTransit/pull/1838

[phatboyg]

Yeah, this looks good. I merged it into develop. So now, if you create a scope and then use that scope to resolve ISendEndpointProvider, it will be scoped and the payload will be present.

[kooshan]

Hi, I happened to have the exact same problem here and I was amazed that the issue has just been resolved and closed for like a couple of hours before!
By the way, does this also work for IRequestClients? I heavily used them though my code and also completely relied on the multi-response feature. While it is possible to add headers using RequestHandle, it does not support multiple responses out of the box, and IRequestClient does not allow for direct context access in order to augment it with custom headers.
I have tried using

cfg.ConfigurePublish(o => {
    o.UseExecute(publishContext => {
       // some scoped service
        var principal = sp.GetRequiredService<IPrincipal>() as ClaimsPrincipal;
        }
   }
})

I don't know what is going wrong but I'm not able to get a principal from the last calling scope (it is not a http controller scope so not HttpContextAccessor here!).
I'm using the last published developed version: 6.2.6-develop.2601
Thank you.

[NooNameR]

@kooshan could you please try:

cfg.ConfigurePublish(o => {
    o.UseExecute(publishContext => {
       // some scoped service
        if(publishContext.TryGetPayload<IServiceProvider>(out var provider))
             var principal = provider.GetRequiredService<IPrincipal>() as ClaimsPrincipal;
        }
   }
})

[kooshan]

@NooNameR
Thank you for your quick response.
But no. That didn't help. I even tried IServiceScope instead of IServiceProvider and TryGetPayload returned false in both cases.
I just ported some of same logic to a MVC controller and checked whether it shows the same behavior in that context. and yes I can't get a scoped service provider from context payload while an instance of HttpContextAccessor in PipeConfigurator body returns a correct instance of principal.

[NooNameR]

so I should change something, I'm on it :)

[phatboyg]

While it is possible to add headers using RequestHandle, it does not support multiple responses out of the box, and IRequestClient does not allow for direct context access in order to augment it with custom headers.

So you understand how the RequestHandle is used to set headers/properties on the SendContext of the request, and those are somehow unusable to solve the problem you're seeing? I realize it's unrelated to the scope issue, or maybe it isn't, but the handle is there specifically for that scenario.

[NooNameR]

it is more about using ScopedSend/Publish transport for ClientFactory :)

[phatboyg]

Oh, got it. Yeah, makes sense.

[kooshan]

@phatboyg Yes I'm aware of that handy accessory, but as I said I diverged from using RequestHandle because it does not have a Send method support two types of responses. IRequestClient provides that but has no signature for tweaking the context. So, it's kind of win/lose situation no matter which one you choose.
Anyway, if adding hearders (Identity headers in my case) has to be applied in all use cases it would be nicer and much cleaner to separate it away in some sort of specification/middleware instead of direct context abuse on every call.

[NooNameR]

@kooshan , should be fixed by: https://github.com/MassTransit/MassTransit/pull/1841

[phatboyg]

@kooshan actually, it does, you just call GetResponse<T1>(false), GetResponse<T2>() which then sends the response, but you are right, the syntax isn't the same but the net result is. I wonder if this is poorly documented. Anyway.

[kooshan]

@NooNameR Thank you. I'll check as soon as the next package hits the web.

[NooNameR]

@kooshan not sure if it be merged, it contains some breaking changes...

[kooshan]

Thank you @NooNameR & @phatboyg. It works great!