Что случилось с InMemoryTransportCache в масстранзите?

Question

Что случилось с InMemoryTransportCache в masstransit, и есть ли сейчас возможность проверить соединение между двумя busControls в памяти?

Answer 1

Похоже, что он был реорганизован.

GetSendEndpoint никогда не разрешит другой экземпляр шины ISendEndpoint, потому что:

var queueName = address.AbsolutePath.Split('/').Last();

игнорирует «Виртуальный хост» в Uri (loopback://localhost/VirtualHost/QueueName)

Существует один тест с двумя экземплярами шины InMemoryDuo_Specs.cs, где он просто передает экземпляр шины для связи.

Наши варианты - поиск экземпляров шины /отправлять конечные точки или одну общую шину с уникальными очередями при тестировании.

Редактировать: Другая проблема для первого решения ConsumeContext. Send внутри обработчика также не будет работать, поэтому вам нужно иметьгрязный способ получить правильный экземпляр.

Answer 2

Вы хотите протестировать, поскольку в памяти было 2 разных хоста, которые обмениваются данными по шине, верно?

Вы можете разрешить службам использовать один и тот же экземпляр IBusControl и регистрировать каждого потребителя с помощью метода ConnectRecieveEndpoint..

Короткий пример, иллюстрирующий использование «ConnectReceiveEndpoint»:

public class TestMessage { public string Text { get; set; } }

[Test]
public async Task Can_publish_and_receive_message()
{
    var consumer1Semaphore = new SemaphoreSlim(0);
    var consumer2Semaphore = new SemaphoreSlim(0);

    IInMemoryHost host = null; 

    var busControl = MassTransit.Bus.Factory.CreateUsingInMemory(inMemoryBusFactoryConfigurator =>
    {
        host = inMemoryBusFactoryConfigurator.Host; 

        inMemoryBusFactoryConfigurator.ReceiveEndpoint("consumer1", ep =>
        {   
            ep.Handler<TestMessage>(context =>
            {
                context.Message.Text.Should().Be("Hi");
                consumer1Semaphore.Release();
                return Task.CompletedTask;
            });
        });
    });

    host.ConnectReceiveEndpoint("consumer2", inMemoryReceiveEndpointConfigurator =>
    {
        inMemoryReceiveEndpointConfigurator.Handler<TestMessage>(context =>
        {
            context.Message.Text.Should().Be("Hi");
            consumer2Semaphore.Release();
            return Task.CompletedTask;
        });

    });

    await busControl.StartAsync();

    await busControl.Publish(new TestMessage{Text = "Hi"});

    (await consumer1Semaphore.WaitAsync(2.Seconds())).Should().BeTrue();
    (await consumer2Semaphore.WaitAsync(2.Seconds())).Should().BeTrue();

    await busControl.StopAsync();
}

Простой подход для тестирования 2 воображаемых сервисов;ServiceA и ServiceB (не размещены в реальной среде).IBusControlFactory, приведенный ниже, используется для возможности замены транспорта служебной шины Azure в пользу транспорта в памяти в тестах.

public interface IBusControlFactory
{
    IBusControl Create(string queueName = null, Action<IReceiveEndpointConfigurator> configurator = null);
}

// Test replacement for a factory that normally creates a Azure Service Bus bus instance
public class Harness : IBusControlFactory, IDisposable
{
    public readonly IBusControl BusControl;

    private IInMemoryHost _host;

    public Harness() => BusControl = MassTransit.Bus.Factory.CreateUsingInMemory(configure: cfg => _host = cfg.Host);

    public static Harness Start() => new Harness();

    public IBusControl Create(string queueName = null, Action<IReceiveEndpointConfigurator> configurator = null)
    {
        if(queueName != null && configurator != null)
            _host.ConnectReceiveEndpoint(queueName, configurator);

        return BusControl;
    }

    public void Dispose() => BusControl.Stop();
}

public class ServiceA
{
    private readonly IBus _bus;

    public ServiceA(IBusControlFactory factory) => _bus = factory.Create();

    public Task SayHi() => _bus.Publish(new TestMessage { Text = "Hi" });
}

public class ServiceB
{
    public ServiceB(IBusControlFactory factory)
    {
        factory.Create("consumer2", ep =>
        {
            ep.Handler<TestMessage>(context =>
            {
                context.Message.Text.Should().Be("Hi");
                return Task.CompletedTask;
            });
        });
    }
}

[Test]
public async Task Can_publish_and_receive_message_take_2()
{
    using (var harness = Harness.Start())
    {
        var serviceA = new ServiceA(harness);
        var serviceB = new ServiceB(harness);

        var spy = new Spy();
        harness.BusControl.ConnectConsumeMessageObserver(spy);

        await serviceA.SayHi();

        await spy.WaitOne(2.Seconds());
    }            
}

public class Spy : IConsumeMessageObserver<TestMessage>
{
    private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0);

    public Task PreConsume(ConsumeContext<TestMessage> context) => Task.CompletedTask;

    public Task PostConsume(ConsumeContext<TestMessage> context)
    {
        _semaphore.Release();
        return Task.CompletedTask;
    }

    public Task ConsumeFault(ConsumeContext<TestMessage> context, Exception exception) => Task.CompletedTask;
    public async Task WaitOne(TimeSpan timeout) { (await _semaphore.WaitAsync(timeout)).Should().BeTrue(); }
}