<commit_before>// Copyright (c) <NAME>. All rights reserved.
// Licensed under the MIT license. See LICENSE.txt file in the project root for full license information.

using System;
using System.Buffers;
using System.Diagnostics;
using System.IO;
using System.IO.Pipelines;
using System.Linq;
using System.Runtime.ExceptionServices;
using System.Threading;
using System.Threading.Tasks;
using Microsoft;
using Microsoft.VisualStudio;
using Microsoft.VisualStudio.Threading;
using Nerdbank.Streams;
using Xunit;
using Xunit.Abstractions;

public class MultiplexingStreamTests : TestBase, IAsyncLifetime
{
    private Stream transport1;
    private Stream transport2;
    private MultiplexingStream mx1;
    private MultiplexingStream mx2;

    public MultiplexingStreamTests(ITestOutputHelper logger)
        : base(logger)
    {
    }

    public async Task InitializeAsync()
    {
        var mx1TraceSource = new TraceSource(nameof(this.mx1), SourceLevels.All);
        var mx2TraceSource = new TraceSource(nameof(this.mx2), SourceLevels.All);

        mx1TraceSource.Listeners.Add(new XunitTraceListener(this.Logger));
        mx2TraceSource.Listeners.Add(new XunitTraceListener(this.Logger));

        Func<string, int, string, TraceSource> traceSourceFactory = (string mxInstanceName, int id, string name) =>
        {
            var traceSource = new TraceSource(mxInstanceName + " channel " + id, SourceLevels.All);
            traceSource.Listeners.Add(new XunitTraceListener(this.Logger));
            return traceSource;
        };

        Func<int, string, TraceSource> mx1TraceSourceFactory = (int id, string name) => traceSourceFactory(nameof(this.mx1), id, name);
        Func<int, string, TraceSource> mx2TraceSourceFactory = (int id, string name) => traceSourceFactory(nameof(this.mx2), id, name);

        (this.transport1, this.transport2) = FullDuplexStream.CreatePair(new System.IO.Pipelines.PipeOptions(pauseWriterThreshold: 2 * 1024 * 1024));
        var mx1 = MultiplexingStream.CreateAsync(this.transport1, new MultiplexingStream.Options { TraceSource = mx1TraceSource, DefaultChannelTraceSourceFactory = mx1TraceSourceFactory }, this.TimeoutToken);
        var mx2 = MultiplexingStream.CreateAsync(this.transport2, new MultiplexingStream.Options { TraceSource = mx2TraceSource, DefaultChannelTraceSourceFactory = mx2TraceSourceFactory }, this.TimeoutToken);
        this.mx1 = await mx1;
        this.mx2 = await mx2;
    }

    public Task DisposeAsync()
    {
        this.mx1?.Dispose();
        this.mx2?.Dispose();
        AssertNoFault(this.mx1);
        AssertNoFault(this.mx2);

        this.mx1?.TraceSource.Listeners.OfType<XunitTraceListener>().SingleOrDefault()?.Dispose();
        this.mx2?.TraceSource.Listeners.OfType<XunitTraceListener>().SingleOrDefault()?.Dispose();

        return TplExtensions.CompletedTask;
    }

    [Fact]
    public async Task Dispose_CancelsOutstandingOperations()
    {
        Task offer = this.mx1.OfferChannelAsync("offer");
        Task accept = this.mx1.AcceptChannelAsync("accept");
        this.mx1.Dispose();
        await Assert.ThrowsAnyAsync<OperationCanceledException>(() => Task.WhenAll(offer, accept)).WithCancellation(this.TimeoutToken);
        Assert.True(offer.IsCanceled);
        Assert.True(accept.IsCanceled);
    }

    [Fact]
    public void Disposal_DisposesTransportStream()
    {
        this.mx1.Dispose();
        Assert.Throws<ObjectDisposedException>(() => this.transport1.Position);
    }

    [Fact]
    public async Task Dispose_DisposesChannels()
    {
        var (channel1, channel2) = await this.EstablishChannelsAsync("A");
        this.mx1.Dispose();
        Assert.True(channel1.IsDisposed);
    }

    [Fact]
    public async Task CreateChannelAsync_ThrowsAfterDisposal()
    {
        this.mx1.Dispose();
        await Assert.ThrowsAsync<ObjectDisposedException>(() => this.mx1.OfferChannelAsync(string.Empty, this.TimeoutToken)).WithCancellation(this.TimeoutToken);
    }

    [Fact]
    public async Task AcceptChannelAsync_ThrowsAfterDisposal()
    {
        this.mx1.Dispose();
        await Assert.ThrowsAsync<ObjectDisposedException>(() => this.mx1.AcceptChannelAsync(string.Empty, this.TimeoutToken)).WithCancellation(this.TimeoutToken);
    }

    [Fact]
    public void Completion_CompletedAfterDisposal()
    {
        this.mx1.Dispose();
        Assert.Equal(TaskStatus.RanToCompletion, this.mx1.Completion.Status);
    }

    [Fact]
    public async Task CreateChannelAsync_NullId()
    {<commit_msg>Add assertions for MultiplexingStream disposal<commit_after>// Copyright (c) <NAME>. All rights reserved.
// Licensed under the MIT license. See LICENSE.txt file in the project root for full license information.

using System;
using System.Buffers;
using System.Diagnostics;
using System.IO;
using System.IO.Pipelines;
using System.Linq;
using System.Runtime.ExceptionServices;
using System.Threading;
using System.Threading.Tasks;
using Microsoft;
using Microsoft.VisualStudio;
using Microsoft.VisualStudio.Threading;
using Nerdbank.Streams;
using Xunit;
using Xunit.Abstractions;

public class MultiplexingStreamTests : TestBase, IAsyncLifetime
{
    private Stream transport1;
    private Stream transport2;
    private MultiplexingStream mx1;
    private MultiplexingStream mx2;

    public MultiplexingStreamTests(ITestOutputHelper logger)
        : base(logger)
    {
    }

    public async Task InitializeAsync()
    {
        var mx1TraceSource = new TraceSource(nameof(this.mx1), SourceLevels.All);
        var mx2TraceSource = new TraceSource(nameof(this.mx2), SourceLevels.All);

        mx1TraceSource.Listeners.Add(new XunitTraceListener(this.Logger));
        mx2TraceSource.Listeners.Add(new XunitTraceListener(this.Logger));

        Func<string, int, string, TraceSource> traceSourceFactory = (string mxInstanceName, int id, string name) =>
        {
            var traceSource = new TraceSource(mxInstanceName + " channel " + id, SourceLevels.All);
            traceSource.Listeners.Add(new XunitTraceListener(this.Logger));
            return traceSource;
        };

        Func<int, string, TraceSource> mx1TraceSourceFactory = (int id, string name) => traceSourceFactory(nameof(this.mx1), id, name);
        Func<int, string, TraceSource> mx2TraceSourceFactory = (int id, string name) => traceSourceFactory(nameof(this.mx2), id, name);

        (this.transport1, this.transport2) = FullDuplexStream.CreatePair(new System.IO.Pipelines.PipeOptions(pauseWriterThreshold: 2 * 1024 * 1024));
        var mx1 = MultiplexingStream.CreateAsync(this.transport1, new MultiplexingStream.Options { TraceSource = mx1TraceSource, DefaultChannelTraceSourceFactory = mx1TraceSourceFactory }, this.TimeoutToken);
        var mx2 = MultiplexingStream.CreateAsync(this.transport2, new MultiplexingStream.Options { TraceSource = mx2TraceSource, DefaultChannelTraceSourceFactory = mx2TraceSourceFactory }, this.TimeoutToken);
        this.mx1 = await mx1;
        this.mx2 = await mx2;
    }

    public Task DisposeAsync()
    {
        this.mx1?.Dispose();
        this.mx2?.Dispose();
        AssertNoFault(this.mx1);
        AssertNoFault(this.mx2);

        this.mx1?.TraceSource.Listeners.OfType<XunitTraceListener>().SingleOrDefault()?.Dispose();
        this.mx2?.TraceSource.Listeners.OfType<XunitTraceListener>().SingleOrDefault()?.Dispose();

        return TplExtensions.CompletedTask;
    }

    [Fact]
    public async Task Dispose_CancelsOutstandingOperations()
    {
        Task offer = this.mx1.OfferChannelAsync("offer");
        Task accept = this.mx1.AcceptChannelAsync("accept");
        this.mx1.Dispose();
        await Assert.ThrowsAnyAsync<OperationCanceledException>(() => Task.WhenAll(offer, accept)).WithCancellation(this.TimeoutToken);
        Assert.True(offer.IsCanceled);
        Assert.True(accept.IsCanceled);
    }

    [Fact]
    public void Disposal_DisposesTransportStream()
    {
        this.mx1.Dispose();
        Assert.Throws<ObjectDisposedException>(() => this.transport1.Position);
    }

    [Fact]
    public async Task Dispose_DisposesChannels()
    {
        var (channel1, channel2) = await this.EstablishChannelsAsync("A");
        Task channel1WriterCompleted = channel1.Input.WaitForWriterCompletionAsync();
        Task channel1ReaderCompleted = channel1.Output.WaitForReaderCompletionAsync();
        this.mx1.Dispose();
        Assert.True(channel1.IsDisposed);
        await channel1.Completion.WithCancellation(this.TimeoutToken);
        await channel1WriterCompleted.WithCancellation(this.TimeoutToken);
        await channel1ReaderCompleted.WithCancellation(this.TimeoutToken);
    }

    [Fact]
    public async Task CreateChannelAsync_ThrowsAfterDisposal()
    {
        this.mx1.Dispose();
        await Assert.ThrowsAsync<ObjectDisposedException>(() => this.mx1.OfferChannelAsync(string.Empty, this.TimeoutToken)).WithCancellation(this.TimeoutToken);
    }

    [Fact]
    public async Task AcceptChannelAsync_ThrowsAfterDisposal()
    {
        this.mx1.Dispose();
        await Assert.ThrowsAsync<ObjectDisposedException>(() => this.mx1.AcceptChannelAsync(string.Empty, this.TimeoutToken)).WithCancellation(this.TimeoutToken);
    }

    [Fact]
    public void Completion_CompletedAfterDisposal()
    {
        this.mx1.Dispose();
        Assert.Equal(TaskStatus.RanToCompletion, this.mx1.Completion.Status);
    }

    [Fact]
    public async Task CreateChannelAsync_NullId()
    {
{'fraction_non_alphanumeric': 0.10336462398481173, 'fraction_numerical': 0.012973315051154942, 'mean_word_length': 3.7696177062374248, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '931533', 'n_tokens_mistral': 2779, 'n_tokens_neox': 2756, 'n_words': 514}