[Testing Platform] IDataConsumer is never invoked

[thomhurst]

Heya, I'm trying to write an extension point, and using the IDataConsumer, but it never seems to be invoked.

Am I doing something wrong or is this a bug?

Repro:

using System.Text.Json;
using Microsoft.Testing.Platform.Builder;
using Microsoft.Testing.Platform.Capabilities.TestFramework;
using Microsoft.Testing.Platform.Extensions;
using Microsoft.Testing.Platform.Extensions.Messages;
using Microsoft.Testing.Platform.Extensions.TestFramework;
using Microsoft.Testing.Platform.Extensions.TestHost;
using Microsoft.Testing.Platform.TestHost;

var builder = await TestApplication.CreateBuilderAsync(args);

var testFramework = new DummyAdapter();

builder.RegisterTestFramework(_ => new TestFrameworkCapabilities(), (_, _) => testFramework);

var myExtension = new MyExtension(testFramework);

builder.TestHost.AddTestApplicationLifecycleCallbacks(_ => myExtension);
builder.TestHost.AddDataConsumer(_ => myExtension);

var app = await builder.BuildAsync();
return await app.RunAsync();

internal class DummyAdapter : ITestFramework, IDataProducer
{
    public string Uid => nameof(DummyAdapter);

    public string Version => string.Empty;

    public string DisplayName => string.Empty;

    public string Description => string.Empty;

    public Type[] DataTypesProduced => [typeof(TestNodeUpdateMessage)];

    public Task<CloseTestSessionResult> CloseTestSessionAsync(CloseTestSessionContext context) => Task.FromResult(new CloseTestSessionResult { IsSuccess = true });

    public Task<CreateTestSessionResult> CreateTestSessionAsync(CreateTestSessionContext context) => Task.FromResult(new CreateTestSessionResult { IsSuccess = true });

    public Task ExecuteRequestAsync(ExecuteRequestContext context)
    {
        context.MessageBus.PublishAsync(this, new TestNodeUpdateMessage(new SessionUid("1"), new TestNode
        {
            Uid = "1",
            DisplayName = "1",
            Properties = new PropertyBag(DiscoveredTestNodeStateProperty.CachedInstance)
        }));
        
        context.MessageBus.PublishAsync(this, new TestNodeUpdateMessage(new SessionUid("1"), new TestNode
        {
            Uid = "2",
            DisplayName = "2",
            Properties = new PropertyBag(DiscoveredTestNodeStateProperty.CachedInstance)
        }));
        
        context.MessageBus.PublishAsync(this, new TestNodeUpdateMessage(new SessionUid("1"), new TestNode
        {
            Uid = "3",
            DisplayName = "3",
            Properties = new PropertyBag(DiscoveredTestNodeStateProperty.CachedInstance)
        }));
        
        context.MessageBus.PublishAsync(this, new TestNodeUpdateMessage(new SessionUid("1"), new TestNode
        {
            Uid = "1",
            DisplayName = "1",
            Properties = new PropertyBag(PassedTestNodeStateProperty.CachedInstance)
        }));
        
        context.MessageBus.PublishAsync(this, new TestNodeUpdateMessage(new SessionUid("1"), new TestNode
        {
            Uid = "2",
            DisplayName = "2",
            Properties = new PropertyBag(new FailedTestNodeStateProperty("Oops"))
        }));
        
        context.MessageBus.PublishAsync(this, new TestNodeUpdateMessage(new SessionUid("1"), new TestNode
        {
            Uid = "3",
            DisplayName = "3",
            Properties = new PropertyBag(SkippedTestNodeStateProperty.CachedInstance)
        }));
     
        context.Complete();
        
        return Task.CompletedTask;
    }

    public Task<bool> IsEnabledAsync() => Task.FromResult(true);
}

public class MyExtension(IExtension extension) : IDataConsumer, ITestApplicationLifecycleCallbacks
{
    public Task ConsumeAsync(IDataProducer dataProducer, IData value, CancellationToken cancellationToken)
    {
        Console.WriteLine($"Consuming data: {JsonSerializer.Serialize(value)}");
        return Task.CompletedTask;
    }

    public Type[] DataTypesConsumed { get; } = [typeof(TestNodeUpdateMessage)];
    
    public Task<bool> IsEnabledAsync()
    {
        return extension.IsEnabledAsync();
    }

    public string Uid => extension.Uid;

    public string Version => extension.Version;

    public string DisplayName => extension.DisplayName;

    public string Description => extension.Description;
    
    public Task BeforeRunAsync(CancellationToken cancellationToken)
    {
        Console.WriteLine("Before Test App");
        return Task.CompletedTask;
    }

    public Task AfterRunAsync(int exitCode, CancellationToken cancellation)
    {
        Console.WriteLine("After Test App");
        return Task.CompletedTask;
    }
}

Stick a breakpoint in ConsumeAsync and you'll see it's never hit, or just see that it doesn't print any console statements.

[Evangelink]

The producer and consumer have the same UID so we don't send the data, see https://github.com/microsoft/testfx/blob/main/src/Platform/Microsoft.Testing.Platform/Messages/ChannelConsumerDataProcessor.cs#L63-L69.

@MarcoRossignoli Do you recall why we did this?

[nohwnd]

(y) Just what I was writing

[thomhurst]

The producer and consumer have the same UID so we don't send the data

Hmm... Can we just let the extension decide that if it wants?

They could just do:

        if (dataProducer.Uid == Uid)
        {
            return;
        }

[thomhurst]

Anyway not a blocker for now - I'll just use a different Uid for my extension. Thanks! 😄

[Evangelink]

Yeah keeping this open so we can discuss, I really don't recall why we added this check. Maybe it was making sense at first and then with the update of the design it's something that no longer makes sense.

[nohwnd]

isn't it something about overhead of sending data via message bus that is unnecessary when the extension is one and the same? I think marco said something like this.

[MarcoRossignoli]

isn't it something about overhead of sending data via message bus that is unnecessary when the extension is one and the same? I think marco said something like this.

Down this line, logically if I know my data I can consume "here". Producer and Consumer should not know each other.

[thomhurst]

isn't it something about overhead of sending data via message bus that is unnecessary when the extension is one and the same? I think marco said something like this.

Down this line, logically if I know my data I can consume "here". Producer and Consumer should not know each other.

Then you have a class that's mixing responsibilities and not following SRP nicely?

[MarcoRossignoli]

Then you have a class that's mixing responsibilities and not following SRP nicely?

You can do it internally in your code that you fully own, you should not relay on the MessageBus, in future we can change the behavior of it, like block the push till all consumer consumed or something like that, why would you pay that cost to elaborate data that you're producing yourselves? Producer/Consumer in expensive and if I have my data I would elaborate the data I've produced next to the producer if I don't need to scale. Or if I need to scale I'd like to have full control on the threading.

[thomhurst]

Producer/Consumer in expensive

If it's extenal / IO then yeah, but isn't this all In-Process? Is it not just passing it from one class to another when extensions are in-process?

[MarcoRossignoli]

If it's extenal / IO then yeah, but isn't this all In-Process? Is it not just passing it from one class to another when extensions are in-process?

The current implementation uses Channels in .NET and a BlockingCollection for .NET Framework so actually you pay the context switch (always for .NET Framework and for .NET Channels use the ThreadPool so could be done inline or on another thread).
Every consumer has got custom "queue" and consumers consume in async isolated and we have a "barrier" at some points to wait the consumptions.

[thomhurst]

Ah okay. Best way to resolve this then would be to mention this logic in the documentation I think?

[Evangelink]

Yeah I think it's better to keep the current behavior and improve documentation.