IAsyncEnumerable handler can be cleaned up early

[BertanAygun]

Per our investigation with @AArnott the following test fails when using MessagePack or Newtonsoft formatters because IAsyncEnumerable formatter created for GetAsync method is cleaned up early when client.DoSomethingAsync() method returns.

What happens is that request id for "server.GetAsync" that is attached to IAsyncEnumerable ends up being same as request id for "client.DoSomethingAsync()" and response of DoSomethingAsync ends up cleaning up the enumerable formatter before enumerable is finalized.

public class AsyncEnumerableTests
{ 
    [Fact]
    public async Task Test()
    {
        var pair = FullDuplexStream.CreatePair();

        var serverHandler = new LengthHeaderMessageHandler(pair.Item1.UsePipe(), new MessagePackFormatter());
        var clientHandler = new LengthHeaderMessageHandler(pair.Item2.UsePipe(), new MessagePackFormatter());

        JsonRpc jsonRpcClient = new JsonRpc(clientHandler);
        JsonRpc jsonRpcServer = new JsonRpc(serverHandler, new Server());

        IServer server = jsonRpcClient.Attach<IServer>();

        jsonRpcServer.StartListening();
        jsonRpcClient.StartListening();

        // uncomment the following code to fix the test.
        // await server.StartAsync();

        await foreach (var s in server.GetAsync(new Client()))
        {
        }
    }
}

public interface IServer
{
    IAsyncEnumerable<string> GetAsync(IClient client);

    Task StartAsync();
}

[RpcMarshalable]
public interface IClient : IDisposable
{
    Task DoSomethingAsync();
}

public class Client : IClient
{
    public Task DoSomethingAsync() { return Task.CompletedTask; }
    public void Dispose() { }
}

public class Server : IServer
{
    public async IAsyncEnumerable<string> GetAsync(IClient client)
    {
        // make a request from server to client, if this ends up with 
        // same request id that called this method IAsyncEnumerable formatter will be cleaned up early
        await client.DoSomethingAsync(); 
        yield return "Hello";
    }

    public Task StartAsync()   { return Task.CompletedTask; }
}

[AArnott]

Sweet. I have a repro in a unit test based on your sample.

[AArnott]

It doesn't require marshalable objects either, BTW.