<commit_before>﻿using Abc.Zebus.Directory;
using Abc.Zebus.Serialization;
using Abc.Zebus.Testing;
using Abc.Zebus.Testing.Extensions;
using Abc.Zebus.Tests.Messages;
using Abc.Zebus.Transport;
using Abc.Zebus.Util;
using Moq;
using NUnit.Framework;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace Abc.Zebus.Tests.Transport
{
    [TestFixture]
    public class ZmqTransportTests
    {
        private const string _environment = "Test";
        private List<ZmqTransport> _transports;

        [SetUp]
        public void Setup()
        {
            _transports = new List<ZmqTransport>();
        }

        [TearDown]
        public void Teardown()
        {
            foreach (var transport in _transports)
            {
                try
                {
                    if (transport.IsListening)
                        transport.Stop(true);
                }
                catch (Exception)
                {
                }
            }
        }

        [Test]
        public void should_not_crash_when_stopping_if_it_was_not_started()
        {
            var configurationMock = new Mock<IZmqTransportConfiguration>();
            configurationMock.SetupGet(x => x.WaitForEndOfStreamAckTimeout).Returns(100.Milliseconds());
            var transport = new ZmqTransport(configurationMock.Object, new ZmqSocketOptions());

            Assert.That(transport.Stop, Throws.Nothing);
        }

        [Test]
        public void should_not_filter_received_messages_when_environment_is_not_specified()
        {
            var transport1 = CreateAndStartZmqTransport(environment: null);

            var transport2ReceivedMessages = new ConcurrentBag<TransportMessage>();
            var transport2 = CreateAndStartZmqTransport(onMessageReceived: transport2ReceivedMessages.Add, environment: "NotTest");
            var transport2Peer = new Peer(transport2.PeerId, transport2.InboundEndPoint);

            var message = new FakeCommand(1).ToTransportMessage();
            transport1.Send(message, new[] { transport2Peer });

            Wait.Until(() => transport2ReceivedMessages.Count >= 1, 500.Milliseconds());
            transport2ReceivedMessages.Single().Id.ShouldEqual(message.Id);
        }

        [Test]
        public void should_not_let_the_outbound_thread_die_if_a_peer_cannot_be_resolved()
        {
            var senderTransport = CreateAndStartZmqTransport(environment: null);

            var receivedMessages = new ConcurrentBag<TransportMessage>();
            var destinationTransport = CreateAndStartZmqTransport(onMessageReceived: receivedMessages.Add, environment: "NotTest");
            var destinationPeer = new Peer(destinationTransport.PeerId, destinationTransport.InboundEndPoint);
            var nonExistingPeer = new Peer(new PeerId("Abc.NonExistingPeer.2"), "tcp://non_existing_peer:1234");

            var message = new FakeCommand(1).ToTransportMessage();
            senderTransport.Send(message, new[] { nonExistingPeer });
            senderTransport.Send(message, new[] { destinationPeer });

            Wait.Until(() => receivedMessages.Count >= 1, 1000.Milliseconds(), "The outbound thread was killed and couldn't connect to the next peer");
        }

        [Test]
        public void should_not_dispatch_messages_received_from_wrong_environment()
        {
            var transport1ReceivedMessages = new ConcurrentBag<TransportMessage>();
            var transport1 = CreateAndStartZmqTransport(onMessageReceived: transport1ReceivedMessages.Add);

            var transport2ReceivedMessages = new ConcurrentBag<TransportMessage>();
            var transport2 = CreateAndStartZmqTransport(onMessageReceived: transport2ReceivedMessages.Add, environment: "NotTest");
            var transport2Peer = new Peer(transport2.PeerId, transport2.InboundEndPoint);

            var message1 = new FakeCommand(1).ToTransportMessage();
            var message2 = new FakeCommand(2).ToTransportMessage();
            transport1.Send(message1, new[] { transport2Peer }); // should not arrive

            Thread.Sleep(500); //:(
            transport2.Configure(transport2Peer.Id, _environment);
            transport1.Send(message2, new[] { transport2Peer }); //should arrive

            Wait.Until(() => transport2ReceivedMessages.Count >= 1, 500.Milliseconds());
            transport2ReceivedMessages.Single().Id.ShouldEqual(message2.Id);
        }

        [Test, Timeout(5000)]
        public void should_send_messages()
        {
            var transport1ReceivedMessages = new ConcurrentBag<TransportMessage>();
            var transport1 = CreateAndStartZmqTransport(onMessageReceived: transport1ReceivedMessages.Add);
            var transport1Peer = new Peer(transport1.PeerId, transport1.InboundEndPoint);

            var transport2ReceivedMessages = new ConcurrentBag<TransportMessage>();
            var transport2 = CreateAndStartZmqTransport(onMessageReceived: transport2ReceivedMessages.Add);
            var transport2Peer = new Peer(transport2.PeerId, transport2.InboundEndPoint);

            var message1 = new FakeCommand(1).ToTransportMessage();
            transport1.Send(message1, new[] { transport2Peer });

            Wait.Until(() => transport2ReceivedMessages.Count == 1, 500.Milliseconds());
            var transport2ReceivedMessage = transport2ReceivedMessages.ExpectedSingle();
            transport2ReceivedMessage.ShouldHaveSamePropertiesAs(message1, "Environment", "WasPersisted");
            transport2ReceivedMessage.Environment.ShouldEqual("Test");
            transport2ReceivedMessage.WasPersisted.ShouldEqual(false);

            var message2 = new FakeCommand(2).ToTransportMessage();
            transport2.Send(message2, new[] { transport1Peer });

            Wait.Until(() => transport1ReceivedMessages.Count == 1, 500.Milliseconds());
            var transport1ReceivedMessage = transport1ReceivedMessages.ExpectedSingle();
            transport1ReceivedMessage.ShouldHaveSamePropertiesAs(message2, "Environment", "WasPersisted");
            transport1ReceivedMessage.Environment.ShouldEqual("Test");
            transport1ReceivedMessage.WasPersisted.ShouldEqual(false);
        }

        [Test, Timeout(5000)]
        public void should_write_WasPersisted_when_requested()
        {
            var sender = CreateAndStartZmqTransport();

            var receivedMessages = new ConcurrentBag<TransportMessage>();
            var receiver = CreateAndStartZmqTransport(onMessageReceived: receivedMessages.Add);
            var receivingPeer = new Peer(receiver.PeerId, receiver.InboundEndPoint);
            var message = new FakeCommand(1).ToTransportMessage();
            var otherMessage = new FakeCommand(2).ToTransportMessage();

            sender.Send(message, new[] { receivingPeer }, new SendContext { PersistentPeerIds = { receivingPeer.Id } });
            sender.Send(otherMessage, new[] { receivingPeer }, new SendContext());

            Wait.Until(() => receivedMessages.Count >= 2, 500.Milliseconds());
            receivedMessages.Single(x => x.Id == message.Id).WasPersisted.ShouldEqual(true);
            receivedMessages.Single(x => x.Id == otherMessage.Id).WasPersisted.ShouldEqual(false);
        }

        [Test, Timeout(5000)]
        public void should_send_message_to_both_persisted_and_non_persisted_peers()
        {
            var sender = CreateAndStartZmqTransport();
            var receivedMessages = new ConcurrentBag<TransportMessage>();

            var receiver1 = CreateAndStartZmqTransport(onMessageReceived: receivedMessages.Add);
            var receivingPeer1 = new Peer(receiver1.PeerId, receiver1.InboundEndPoint);

            var receiver2 = CreateAndStartZmqTransport(onMessageReceived: receivedMessages.Add);
            var receivingPeer2 = new Peer(receiver2.PeerId, receiver2.InboundEndPoint);<commit_msg>Add missing tests to ZmqTransport<commit_after>﻿using Abc.Zebus.Directory;
using Abc.Zebus.Serialization;
using Abc.Zebus.Testing;
using Abc.Zebus.Testing.Extensions;
using Abc.Zebus.Tests.Messages;
using Abc.Zebus.Transport;
using Abc.Zebus.Util;
using Moq;
using NUnit.Framework;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace Abc.Zebus.Tests.Transport
{
    [TestFixture]
    public class ZmqTransportTests
    {
        private const string _environment = "Test";
        private List<ZmqTransport> _transports;

        [SetUp]
        public void Setup()
        {
            _transports = new List<ZmqTransport>();
        }

        [TearDown]
        public void Teardown()
        {
            foreach (var transport in _transports)
            {
                try
                {
                    if (transport.IsListening)
                        transport.Stop(true);
                }
                catch (Exception)
                {
                }
            }
        }

        [Test]
        public void should_not_crash_when_stopping_if_it_was_not_started()
        {
            var configurationMock = new Mock<IZmqTransportConfiguration>();
            configurationMock.SetupGet(x => x.WaitForEndOfStreamAckTimeout).Returns(100.Milliseconds());
            var transport = new ZmqTransport(configurationMock.Object, new ZmqSocketOptions());

            Assert.That(transport.Stop, Throws.Nothing);
        }

        [Test]
        public void should_not_filter_received_messages_when_environment_is_not_specified()
        {
            var transport1 = CreateAndStartZmqTransport(environment: null);

            var transport2ReceivedMessages = new ConcurrentBag<TransportMessage>();
            var transport2 = CreateAndStartZmqTransport(onMessageReceived: transport2ReceivedMessages.Add, environment: "NotTest");
            var transport2Peer = new Peer(transport2.PeerId, transport2.InboundEndPoint);

            var message = new FakeCommand(1).ToTransportMessage();
            transport1.Send(message, new[] { transport2Peer });

            Wait.Until(() => transport2ReceivedMessages.Count >= 1, 500.Milliseconds());
            transport2ReceivedMessages.Single().Id.ShouldEqual(message.Id);
        }

        [Test]
        public void should_not_let_the_outbound_thread_die_if_a_peer_cannot_be_resolved()
        {
            var senderTransport = CreateAndStartZmqTransport(environment: null);

            var receivedMessages = new ConcurrentBag<TransportMessage>();
            var destinationTransport = CreateAndStartZmqTransport(onMessageReceived: receivedMessages.Add, environment: "NotTest");
            var destinationPeer = new Peer(destinationTransport.PeerId, destinationTransport.InboundEndPoint);
            var nonExistingPeer = new Peer(new PeerId("Abc.NonExistingPeer.2"), "tcp://non_existing_peer:1234");

            var message = new FakeCommand(1).ToTransportMessage();
            senderTransport.Send(message, new[] { nonExistingPeer });
            senderTransport.Send(message, new[] { destinationPeer });

            Wait.Until(() => receivedMessages.Count >= 1, 1000.Milliseconds(), "The outbound thread was killed and couldn't connect to the next peer");
        }

        [Test]
        public void should_not_dispatch_messages_received_from_wrong_environment()
        {
            var transport1ReceivedMessages = new ConcurrentBag<TransportMessage>();
            var transport1 = CreateAndStartZmqTransport(onMessageReceived: transport1ReceivedMessages.Add);

            var transport2ReceivedMessages = new ConcurrentBag<TransportMessage>();
            var transport2 = CreateAndStartZmqTransport(onMessageReceived: transport2ReceivedMessages.Add, environment: "NotTest");
            var transport2Peer = new Peer(transport2.PeerId, transport2.InboundEndPoint);

            var message1 = new FakeCommand(1).ToTransportMessage();
            var message2 = new FakeCommand(2).ToTransportMessage();
            transport1.Send(message1, new[] { transport2Peer }); // should not arrive

            Thread.Sleep(500); //:(
            transport2.Configure(transport2Peer.Id, _environment);
            transport1.Send(message2, new[] { transport2Peer }); //should arrive

            Wait.Until(() => transport2ReceivedMessages.Count >= 1, 500.Milliseconds());
            transport2ReceivedMessages.Single().Id.ShouldEqual(message2.Id);
        }

        [Test, Timeout(5000)]
        public void should_send_messages()
        {
            var transport1ReceivedMessages = new ConcurrentBag<TransportMessage>();
            var transport1 = CreateAndStartZmqTransport(onMessageReceived: transport1ReceivedMessages.Add);
            var transport1Peer = new Peer(transport1.PeerId, transport1.InboundEndPoint);

            var transport2ReceivedMessages = new ConcurrentBag<TransportMessage>();
            var transport2 = CreateAndStartZmqTransport(onMessageReceived: transport2ReceivedMessages.Add);
            var transport2Peer = new Peer(transport2.PeerId, transport2.InboundEndPoint);

            var message1 = new FakeCommand(1).ToTransportMessage();
            transport1.Send(message1, new[] { transport2Peer });

            Wait.Until(() => transport2ReceivedMessages.Count == 1, 500.Milliseconds());
            var transport2ReceivedMessage = transport2ReceivedMessages.ExpectedSingle();
            transport2ReceivedMessage.ShouldHaveSamePropertiesAs(message1, "Environment", "WasPersisted");
            transport2ReceivedMessage.Environment.ShouldEqual("Test");
            transport2ReceivedMessage.WasPersisted.ShouldEqual(false);

            var message2 = new FakeCommand(2).ToTransportMessage();
            transport2.Send(message2, new[] { transport1Peer });

            Wait.Until(() => transport1ReceivedMessages.Count == 1, 500.Milliseconds());
            var transport1ReceivedMessage = transport1ReceivedMessages.ExpectedSingle();
            transport1ReceivedMessage.ShouldHaveSamePropertiesAs(message2, "Environment", "WasPersisted");
            transport1ReceivedMessage.Environment.ShouldEqual("Test");
            transport1ReceivedMessage.WasPersisted.ShouldEqual(false);
        }

        [Test]
        public void should_send_message_to_peer_and_persistence()
        {
            // standard case: the message is forwarded to the persistence through SendContext.PersistencePeer

            var senderTransport = CreateAndStartZmqTransport();

            var receiverMessages = new ConcurrentBag<TransportMessage>();
            var receiverTransport = CreateAndStartZmqTransport(onMessageReceived: receiverMessages.Add);
            var receiverPeer = new Peer(receiverTransport.PeerId, receiverTransport.InboundEndPoint);

            var persistenceMessages = new ConcurrentBag<TransportMessage>();
            var persistenceTransport = CreateAndStartZmqTransport(onMessageReceived: persistenceMessages.Add);
            var persistencePeer = new Peer(persistenceTransport.PeerId, persistenceTransport.InboundEndPoint);

            var message = new FakeCommand(999).ToTransportMessage();
            senderTransport.Send(message, new[] { receiverPeer }, new SendContext { PersistentPeerIds = { receiverPeer.Id }, PersistencePeer = persistencePeer });

            Wait.Until(() => receiverMessages.Count == 1, 500.Milliseconds());
            var messageFromReceiver = receiverMessages.ExpectedSingle();
            messageFromReceiver.ShouldHaveSamePropertiesAs(message, "Environment", "WasPersisted");
            messageFromReceiver.Environment.ShouldEqual("Test");
            messageFromReceiver.WasPersisted.ShouldEqual(true);

            Wait.Until(() => persistenceMessages.Count == 1, 500.Milliseconds());
            var messageFromPersistence = persistenceMessages.ExpectedSingle();
            messageFromPersistence.ShouldHaveSamePropertiesAs(message, "Environment", "WasPersisted", "PersistentPeerIds", "IsPersistTransportMessage");
            messageFromPersistence.Environment.ShouldEqual("Test");
            messageFromPersistence.PersistentPeerIds.ShouldBeEquivalentTo(new[] { receiverPeer.Id });
        }

        [Test]
        public void should_send_persist_transport_message_to_persistence()
        {
            // edge case: the message is directly forwarded to the persistence

            var senderTransport = CreateAndStartZmqTransport();

            var receiverPeerId = new PeerId("Abc.Receiver.123");

            var persistenceMessages = new ConcurrentBag<TransportMessage>();
            var persistenceTransport = CreateAndStartZmqTransport(onMessageReceived: persistenceMessages.Add);
            var persistencePeer = new Peer(persistenceTransport.PeerId, persistenceTransport.InboundEndPoint);

            var message = new FakeCommand(999).ToTransportMessage().ToPersistTransportMessage(receiverPeerId);
            senderTransport.Send(message, new[] { persistencePeer });

            Wait.Until(() => persistenceMessages.Count == 1, 500.Milliseconds());
            var messageFromPersistence = persistenceMessages.ExpectedSingle();
            messageFromPersistence.ShouldHaveSamePropertiesAs(message, "Environment", "WasPersisted");
            messageFromPersistence.Environment.ShouldEqual("Test");
            messageFromPersistence.PersistentPeerIds.ShouldBeEquivalentTo(new[] { receiverPeerId });
        }

        [Test, Timeout(5000)]
        public void should_write_WasPersisted_when_requested()
        {
            var sender = CreateAndStartZmqTransport();

            var receivedMessages = new ConcurrentBag<TransportMessage>();
            var receiver = CreateAndStartZmqTransport(onMessageReceived: receivedMessages.Add);
            var receivingPeer = new Peer(receiver.PeerId, receiver.InboundEndPoint);
            var message = new FakeCommand(1).ToTransportMessage();
            var otherMessage = new FakeCommand(2).ToTransportMessage();

            sender.Send(message, new[] { receivingPeer }, new SendContext { PersistentPeerIds = { receivingPeer.Id } });
            sender.Send(otherMessage, new[] { receivingPeer }, new SendContext());

            Wait.Until(() => receivedMessages.Count >= 2, 500.Milliseconds());
            receivedMessages.Single(x => x.Id == message.Id).WasPersisted.ShouldEqual(true);
            receivedMessages.Single(x => x.Id == otherMessage.Id).WasPersisted.ShouldEqual(false);
        }

        [Test, Timeout(5000)]
        public void should_send_message_to_both_persisted_and_non_persisted_peers()
        {
            var sender = CreateAndStartZmqTransport();
            var receivedMessages = new ConcurrentBag<TransportMessage>();

            var receiver1 = CreateAndStartZmqTransport(onMessageReceived: receivedMessages.Add);
            var receivingPeer1 = new Peer(receiver1.PeerId, receiver1.InboundEndPoint);

            var receiver2 = CreateAndStartZmqTransport(onMessageReceived: receivedMessages.Add);
            var receivingPeer2 = new Peer(receiver2.PeerId, receiver2.InboundEndPoint);
{'fraction_non_alphanumeric': 0.09154491762511657, 'fraction_numerical': 0.014661693088799088, 'mean_word_length': 3.0732221987761132, 'pattern_counts': {'":': 0, '<': 28, '<?xml version=': 0, '>': 57, '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': '409436', 'n_tokens_mistral': 4900, 'n_tokens_neox': 4662, 'n_words': 905}