<commit_before>// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.

using System.Collections.Generic;
using System.Linq;
using Xunit;

namespace System.Threading.Tasks.Dataflow.Tests
{
    public class DataflowBlockExtensionsTests
    {
        [Fact]
        public void TestDataflowMessageHeader()
        {
            Assert.Throws<ArgumentException>(() => new DataflowMessageHeader(0));

            Assert.False(new DataflowMessageHeader().IsValid);
            Assert.True(new DataflowMessageHeader(1).IsValid);
            Assert.True(new DataflowMessageHeader(-1).IsValid);

            Assert.Equal(expected: 42, actual: new DataflowMessageHeader(42).Id);

            Assert.True(new DataflowMessageHeader(1).Equals(new DataflowMessageHeader(1)));
            Assert.False(new DataflowMessageHeader(1).Equals(new DataflowMessageHeader(2)));

            Assert.True(new DataflowMessageHeader(1).Equals((object)new DataflowMessageHeader(1)));
            Assert.False(new DataflowMessageHeader(1).Equals((object)new DataflowMessageHeader(2)));
            Assert.False(new DataflowMessageHeader(1).Equals("string"));

            Assert.True(new DataflowMessageHeader(100) == new DataflowMessageHeader(100));
            Assert.False(new DataflowMessageHeader(100) == new DataflowMessageHeader(101));
            Assert.True(new DataflowMessageHeader(100) != new DataflowMessageHeader(101));
            Assert.False(new DataflowMessageHeader(100) != new DataflowMessageHeader(100));

            Assert.True(new DataflowMessageHeader(42).GetHashCode() == 42);
        }

        [Fact]
        public void TestNullTarget_NonNull()
        {
            Assert.NotNull(DataflowBlock.NullTarget<int>());
            Assert.NotNull(DataflowBlock.NullTarget<object>());
            Assert.False(DataflowBlock.NullTarget<int>().Completion.IsCompleted);
        }

        [Fact]
        public async Task TestNullTarget_OfferMessage()
        {
            DataflowTestHelpers.TestOfferMessage_ArgumentValidation(DataflowBlock.NullTarget<int>());
            DataflowTestHelpers.TestOfferMessage_AcceptsDataDirectly(DataflowBlock.NullTarget<string>());
            await DataflowTestHelpers.TestOfferMessage_AcceptsViaLinking(DataflowBlock.NullTarget<double>());

            // Test OfferMessage(consumeToAccept: false)
            Assert.Equal(
                expected: DataflowMessageStatus.Accepted,
                actual: DataflowBlock.NullTarget<double>().OfferMessage(new DataflowMessageHeader(1), 3.14, null, consumeToAccept: false));

            // Test OfferMessage(consumeToAccept: true)
            long consumedId = -1;
            DataflowBlock.NullTarget<int>().OfferMessage(new DataflowMessageHeader(42), 84, new DelegatePropagator<int, int>()
            {
                ConsumeMessageDelegate = delegate(DataflowMessageHeader messageHeader, ITargetBlock<int> target, out bool messageConsumed) {
                    consumedId = messageHeader.Id;
                    messageConsumed = true;
                    return 0;
                }
            }, consumeToAccept: true);
            Assert.Equal(expected: 42, actual: consumedId);

            // Test bad source
            Assert.Throws<InvalidOperationException>(() => {
                var target = DataflowBlock.NullTarget<int>();
                DataflowBlock.NullTarget<int>().OfferMessage(new DataflowMessageHeader(42), 84, new DelegatePropagator<int, int>()
                {
                    ConsumeMessageDelegate = delegate(DataflowMessageHeader _, ITargetBlock<int> __, out bool ___) {
                        throw new InvalidOperationException();
                    }
                }, consumeToAccept: true);
                Assert.True(target.Post(42));
            });

            // Test message no longer available
            var stingySource = new DelegatePropagator<int, int>
            {
                ConsumeMessageDelegate = (DataflowMessageHeader messageHeader, ITargetBlock<int> target, out bool messageConsumed) => {
                    messageConsumed = false;
                    return 0;
                }
            };
            Assert.Equal(
                expected: DataflowMessageStatus.NotAvailable,
                actual: DataflowBlock.NullTarget<int>().OfferMessage(
                    new DataflowMessageHeader(1), 1, stingySource, consumeToAccept: true));
        }

        [Fact]
        public void TestNullTarget_Completion()
        {
            var target = DataflowBlock.NullTarget<object>();

            Assert.NotNull(target.Completion);
            Assert.Equal(target.Completion, target.Completion);

            target.Complete();
            target.Fault(new Exception());
            Assert.False(target.Completion.IsCompleted);

            Assert.NotSame(
                DataflowBlock.NullTarget<object>().Completion,
                DataflowBlock.NullTarget<object>().Completion);
            Assert.NotSame(
                DataflowBlock.NullTarget<int>().Completion,
                DataflowBlock.NullTarget<int>().Completion);
        }

        [Fact]
        [OuterLoop] // finalizer/GC interactions
        public void TestNullTarget_CompletionNoCaching()
        {
            // Make sure that the Completion task returned by a NullTarget
            // is not cached across all NullTargets.  Since it'll never complete,
            // that would be a potentially huge memory leak.

            var state = new object();
            var wro = new WeakReference<object>(state);
            DataflowBlock.NullTarget<int>().Completion.ContinueWith(delegate { }, state);
            state = null;

            GC.Collect();
            GC.WaitForPendingFinalizers();
            GC.Collect();

            Assert.False(wro.TryGetTarget(out state));
        }

        [Fact]
        public void TestPost_ArgumentValidation()
        {
            Assert.Throws<ArgumentNullException>(() => ((ITargetBlock<int>)null).Post(42));
        }

        [Fact]
        public void TestAsObservableAndAsObserver_ArgumentValidation()
        {
            Assert.Throws<ArgumentNullException>(() => ((ISourceBlock<int>)null).AsObservable());
            Assert.Throws<ArgumentNullException>(() => ((ITargetBlock<int>)null).AsObserver());
            Assert.Throws<ArgumentNullException>(() => new BufferBlock<int>().AsObservable().Subscribe(null));
        }

        [Fact]
        public async Task TestAsObservableAndAsObserver_MixedTargets()
        {
            var tcs = new TaskCompletionSource<int>();

            var source = new BufferBlock<int>();
            source.AsObservable().Subscribe(new DelegateObserver<int>
            {
                OnNextDelegate = i => tcs.TrySetResult(i)
            });

            source.Post(42);
            Assert.Equal(expected: 42, actual: await tcs.Task);
        }
<commit_msg>Fix failing test related to GC liveness change<commit_after>// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.

using System.Collections.Generic;
using System.Linq;
using System.Runtime.CompilerServices;
using Xunit;

namespace System.Threading.Tasks.Dataflow.Tests
{
    public class DataflowBlockExtensionsTests
    {
        [Fact]
        public void TestDataflowMessageHeader()
        {
            Assert.Throws<ArgumentException>(() => new DataflowMessageHeader(0));

            Assert.False(new DataflowMessageHeader().IsValid);
            Assert.True(new DataflowMessageHeader(1).IsValid);
            Assert.True(new DataflowMessageHeader(-1).IsValid);

            Assert.Equal(expected: 42, actual: new DataflowMessageHeader(42).Id);

            Assert.True(new DataflowMessageHeader(1).Equals(new DataflowMessageHeader(1)));
            Assert.False(new DataflowMessageHeader(1).Equals(new DataflowMessageHeader(2)));

            Assert.True(new DataflowMessageHeader(1).Equals((object)new DataflowMessageHeader(1)));
            Assert.False(new DataflowMessageHeader(1).Equals((object)new DataflowMessageHeader(2)));
            Assert.False(new DataflowMessageHeader(1).Equals("string"));

            Assert.True(new DataflowMessageHeader(100) == new DataflowMessageHeader(100));
            Assert.False(new DataflowMessageHeader(100) == new DataflowMessageHeader(101));
            Assert.True(new DataflowMessageHeader(100) != new DataflowMessageHeader(101));
            Assert.False(new DataflowMessageHeader(100) != new DataflowMessageHeader(100));

            Assert.True(new DataflowMessageHeader(42).GetHashCode() == 42);
        }

        [Fact]
        public void TestNullTarget_NonNull()
        {
            Assert.NotNull(DataflowBlock.NullTarget<int>());
            Assert.NotNull(DataflowBlock.NullTarget<object>());
            Assert.False(DataflowBlock.NullTarget<int>().Completion.IsCompleted);
        }

        [Fact]
        public async Task TestNullTarget_OfferMessage()
        {
            DataflowTestHelpers.TestOfferMessage_ArgumentValidation(DataflowBlock.NullTarget<int>());
            DataflowTestHelpers.TestOfferMessage_AcceptsDataDirectly(DataflowBlock.NullTarget<string>());
            await DataflowTestHelpers.TestOfferMessage_AcceptsViaLinking(DataflowBlock.NullTarget<double>());

            // Test OfferMessage(consumeToAccept: false)
            Assert.Equal(
                expected: DataflowMessageStatus.Accepted,
                actual: DataflowBlock.NullTarget<double>().OfferMessage(new DataflowMessageHeader(1), 3.14, null, consumeToAccept: false));

            // Test OfferMessage(consumeToAccept: true)
            long consumedId = -1;
            DataflowBlock.NullTarget<int>().OfferMessage(new DataflowMessageHeader(42), 84, new DelegatePropagator<int, int>()
            {
                ConsumeMessageDelegate = delegate(DataflowMessageHeader messageHeader, ITargetBlock<int> target, out bool messageConsumed) {
                    consumedId = messageHeader.Id;
                    messageConsumed = true;
                    return 0;
                }
            }, consumeToAccept: true);
            Assert.Equal(expected: 42, actual: consumedId);

            // Test bad source
            Assert.Throws<InvalidOperationException>(() => {
                var target = DataflowBlock.NullTarget<int>();
                DataflowBlock.NullTarget<int>().OfferMessage(new DataflowMessageHeader(42), 84, new DelegatePropagator<int, int>()
                {
                    ConsumeMessageDelegate = delegate(DataflowMessageHeader _, ITargetBlock<int> __, out bool ___) {
                        throw new InvalidOperationException();
                    }
                }, consumeToAccept: true);
                Assert.True(target.Post(42));
            });

            // Test message no longer available
            var stingySource = new DelegatePropagator<int, int>
            {
                ConsumeMessageDelegate = (DataflowMessageHeader messageHeader, ITargetBlock<int> target, out bool messageConsumed) => {
                    messageConsumed = false;
                    return 0;
                }
            };
            Assert.Equal(
                expected: DataflowMessageStatus.NotAvailable,
                actual: DataflowBlock.NullTarget<int>().OfferMessage(
                    new DataflowMessageHeader(1), 1, stingySource, consumeToAccept: true));
        }

        [Fact]
        public void TestNullTarget_Completion()
        {
            var target = DataflowBlock.NullTarget<object>();

            Assert.NotNull(target.Completion);
            Assert.Equal(target.Completion, target.Completion);

            target.Complete();
            target.Fault(new Exception());
            Assert.False(target.Completion.IsCompleted);

            Assert.NotSame(
                DataflowBlock.NullTarget<object>().Completion,
                DataflowBlock.NullTarget<object>().Completion);
            Assert.NotSame(
                DataflowBlock.NullTarget<int>().Completion,
                DataflowBlock.NullTarget<int>().Completion);
        }

        [Fact]
        [OuterLoop] // finalizer/GC interactions
        public void TestNullTarget_CompletionNoCaching()
        {
            // Make sure that the Completion task returned by a NullTarget
            // is not cached across all NullTargets.  Since it'll never complete,
            // that would be a potentially huge memory leak.
            var wro = CreateWeakReferenceToObjectReferencedByNullTargetContinuation();

            GC.Collect();
            GC.WaitForPendingFinalizers();
            GC.Collect();

            object state;
            Assert.False(wro.TryGetTarget(out state));
        }

        [MethodImpl(MethodImplOptions.NoInlining)]
        private static WeakReference<object> CreateWeakReferenceToObjectReferencedByNullTargetContinuation()
        {
            var state = new object();
            var wro = new WeakReference<object>(state);
            DataflowBlock.NullTarget<int>().Completion.ContinueWith(delegate { }, state);
            return wro;
        }

        [Fact]
        public void TestPost_ArgumentValidation()
        {
            Assert.Throws<ArgumentNullException>(() => ((ITargetBlock<int>)null).Post(42));
        }

        [Fact]
        public void TestAsObservableAndAsObserver_ArgumentValidation()
        {
            Assert.Throws<ArgumentNullException>(() => ((ISourceBlock<int>)null).AsObservable());
            Assert.Throws<ArgumentNullException>(() => ((ITargetBlock<int>)null).AsObserver());
            Assert.Throws<ArgumentNullException>(() => new BufferBlock<int>().AsObservable().Subscribe(null));
        }

        [Fact]
        public async Task TestAsObservableAndAsObserver_MixedTargets()
        {
            var tcs = new TaskCompletionSource<int>();

            var source = new BufferBlock<int>();
            source.AsObservable().Subscribe(new DelegateObserver<int>
            {
                OnNextDelegate = i => tcs.TrySetResult(i)
            });

            source.Post(42);
            Assert.Equal(expected: 42, actual: await tcs.Task);
        }

{'fraction_non_alphanumeric': 0.10050529521700007, 'fraction_numerical': 0.009829030248494497, 'mean_word_length': 2.340578034682081, 'pattern_counts': {'":': 0, '<': 78, '<?xml version=': 0, '>': 94, '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': '1569159', 'n_tokens_mistral': 3720, 'n_tokens_neox': 3519, 'n_words': 673}