public FlowStream(ulong id, string signature, Peer peer, BaseRtmfpProtocol handler, Session band,FlowWriter localFlow)
     : base(id, signature, _Name, peer, handler, band, localFlow)
 {
     if (handler.StreamsManager?.StreamsByType.ContainsKey(StreamTypes.ST_IN_NET_RTMFP)??false)
     _publisher = handler.StreamsManager.StreamsByType[StreamTypes.ST_IN_NET_RTMFP].Select(x => x.Value as InNetRtmfpStream).SingleOrDefault(x=>x.PublisherId==StreamId);
     //_publication = handler.Publications.Values.SingleOrDefault(x => x.PublisherId == _index);
 }

    /// <summary>
    /// Constructor
    /// </summary>
    public ChooseServiceProviderForm(Peer peerObject, ConnectWizard connectionWizard)
    {
        //
        // Required for Windows Form Designer support
        //
        InitializeComponent();
        peer = peerObject;
        this.connectionWizard = connectionWizard;
        this.Text = connectionWizard.SampleName + " - " + this.Text;
        // Fill up our listbox with the service providers
        ServiceProviderInformation[] serviceProviders = peer.GetServiceProviders(false);
        foreach (ServiceProviderInformation info in serviceProviders)
            lstSP.Items.Add(info);

        txtUser.Text = null;
        //Get the default username from the registry if it exists
        Microsoft.Win32.RegistryKey regKey = Microsoft.Win32.Registry.CurrentUser.OpenSubKey("Software\\Microsoft\\DirectX\\SDK\\csDPlay");
        if (regKey != null) {
            try {
                txtUser.Text = (string)regKey.GetValue("DirectPlayUserName", null);
                lstSP.SelectedIndex = (int)regKey.GetValue("DirectPlaySPIndex", 0);
                regKey.Close();
            }
            catch {
                txtUser.Text = null;
                lstSP.SelectedIndex = 0;
            }
        }
        else
            lstSP.SelectedIndex = 0;

        if ((txtUser.Text == null) || (txtUser.Text == "")) {
            txtUser.Text = SystemInformation.UserName;
        }
    }

 /// <summary>
 ///     Create new instance of PeerConnectionFailedEventArgs for peer from given torrent.
 /// </summary>
 /// <param name="manager"></param>
 /// <param name="peer"></param>
 /// <param name="direction">Which direction the connection attempt was</param>
 /// <param name="message">Message associated with the failure</param>
 public PeerConnectionFailedEventArgs(TorrentManager manager, Peer peer, Direction direction, string message)
     : base(manager)
 {
     Peer = peer;
     ConnectionDirection = direction;
     Message = message;
 }

        public Task SendGetHeaders(Peer peer)
        {
            if (!IsStarted)
                return Task.CompletedTask;

            var targetChainLocal = this.coreDaemon.TargetChain;
            if (targetChainLocal == null)
                return Task.CompletedTask;

            var blockLocatorHashes = CalculateBlockLocatorHashes(targetChainLocal.Blocks);

            // remove an existing request for this peer if it's stale
            var now = DateTimeOffset.Now;
            DateTimeOffset requestTime;
            if (this.headersRequestsByPeer.TryGetValue(peer, out requestTime)
                && (now - requestTime) > STALE_REQUEST_TIME)
            {
                this.headersRequestsByPeer.TryRemove(peer, out requestTime);
            }

            // determine if a new request can be made
            if (this.headersRequestsByPeer.TryAdd(peer, now))
                // send out the request for headers
                return peer.Sender.SendGetHeaders(blockLocatorHashes, hashStop: UInt256.Zero);
            else
                return Task.CompletedTask;
        }

        public void should_resolve_peer_using_basic_round_robin_for_different_commands()
        {
            // Arrange
            var resolver = new RoundRobinPeerSelector();
            
            var command1 = new FakeCommand(42);
            var command1Peer1 = new Peer(new PeerId("command1peer1"), "endpoint1");
            var command1Peer2 = new Peer(new PeerId("command1peer2"), "endpoint2");
            var command1HandlingPeer = new[] { command1Peer1, command1Peer2 };

            var command2 = new FakeInfrastructureCommand();
            var command2Peer1 = new Peer(new PeerId("command2peer1"), "endpoint1");
            var command2Peer2 = new Peer(new PeerId("command2peer2"), "endpoint2");
            var command2HandlingPeer = new[] { command2Peer1, command2Peer2 };

            // Act - Assert
            var resolvedPeer = resolver.GetTargetPeer(command1, command1HandlingPeer);
            resolvedPeer.ShouldEqual(command1Peer1);

            resolvedPeer = resolver.GetTargetPeer(command1, command1HandlingPeer);
            resolvedPeer.ShouldEqual(command1Peer2);

            resolvedPeer = resolver.GetTargetPeer(command2, command2HandlingPeer);
            resolvedPeer.ShouldEqual(command2Peer1);

            resolvedPeer = resolver.GetTargetPeer(command1, command1HandlingPeer);
            resolvedPeer.ShouldEqual(command1Peer1);

            resolvedPeer = resolver.GetTargetPeer(command2, command2HandlingPeer);
            resolvedPeer.ShouldEqual(command2Peer2);

            resolvedPeer = resolver.GetTargetPeer(command2, command2HandlingPeer);
            resolvedPeer.ShouldEqual(command2Peer1);
        }

        internal ConnectionPool(Peer peer, int preAllocateAmount)
        {
            #if DEBUG
            if (peer == null)
            {
                throw new ArgumentNullException("peer");
            }
            #endif

            this.peer = peer;

            Connection connection;

            for (var i = 0; i < preAllocateAmount; ++i)
            {
                if (TryAllocate(out connection))
                {
                    if (!TryPush(connection))
                    {
                        //TODO: Error
                    }
                }
                else
                {
                    //TODO: Error
                    break;
                }
            }
        }

        /// <summary>
        /// Gets the peer metadata engine.
        /// </summary>
        /// <param name="engine">The engine.</param>
        /// <param name="peer">The peer.</param>
        /// <returns></returns>
        private IPersistenceEngine GetPeerMetadataEngine(SyncEngine engine, Peer peer)
        {
            if (engine.SecondaryMetadataEngine == null || peer == Peer.Client)
                return engine.MetadataEngine;

            return engine.SecondaryMetadataEngine;
        }

    /// <summary>
    /// Constructor
    /// </summary>
    public CreateJoinForm(Peer peerObject, Address addressObject, ConnectWizard connectionWizard)
    {
        //
        // Required for Windows Form Designer support
        //
        InitializeComponent();
        peer = peerObject;
        this.connectionWizard = connectionWizard;
        this.Text = connectionWizard.SampleName + " - " + this.Text;
        deviceAddress = addressObject;

        //Set up the event handlers
        peer.FindHostResponse += new FindHostResponseEventHandler(FindHostResponseMessage);
        peer.ConnectComplete += new ConnectCompleteEventHandler(ConnectResult);
        peer.AsyncOperationComplete += new AsyncOperationCompleteEventHandler(CancelAsync);

        //Set up our timer
        updateListTimer = new System.Timers.Timer(300); // A 300 ms interval
        updateListTimer.Elapsed += new System.Timers.ElapsedEventHandler(this.UpdateTimer);
        updateListTimer.SynchronizingObject = this;
        updateListTimer.Start();
        //Set up our connect timer
        connectTimer = new System.Timers.Timer(100); // A 100ms interval
        connectTimer.Elapsed += new System.Timers.ElapsedEventHandler(this.ConnectTimer);
        connectTimer.SynchronizingObject = this;
        // Set up our connect event
        connectEvent = new ManualResetEvent(false);
    }

 public void Start(Peer peer, uint publisherId, FlowWriter controller)
 {
     if (PublisherId != 0)
     {
         if (controller != null)
         {
             controller.WriteStatusResponse("Publish.BadName", Name + "is already published");
         }
     }
     PublisherId = publisherId;
    
     string error;
     if (!peer.OnPublish(this, out error))
     {
         if (String.IsNullOrEmpty(error)) error = "Not allowed to publish " + Name;
     }
     _publisher = peer;
     _controller = controller;
     _firstKeyFrame = false;
     foreach (var listener in Listeners)
     {
         listener.Value.StartPublishing(Name);
     }
     Flush();
     if (controller != null)
     {
         controller.WriteStatusResponse("Publish.Start", Name + "is now published");
     }
 }

 private void UpdatePeerSubscription(Peer peer, BindingKey subscription, UpdateAction action)
 {
     if (subscription.IsEmpty)
         UpdatePeersMatchingAllMessages(peer, action);
     else
         _rootNode.Update(peer, subscription, action);
 }

 public PeerDescriptor(PeerId id, string endPoint, bool isPersistent, bool isUp, bool isResponding, DateTime timestampUtc, params Subscription[] subscriptions)
 {
     Peer = new Peer(id, endPoint, isUp, isResponding);
     Subscriptions = subscriptions;
     IsPersistent = isPersistent;
     TimestampUtc = timestampUtc;
 }

        private void UpdateCore(Peer peer, Subscription subscription, bool isAddOrUpdate)
        {
            if (subscription.IsMatchingAllMessages)
            {
                var list = _peersHandlingAllMessages
                    .Where(i => i.Id != peer.Id)
                    .ToList();

                if (isAddOrUpdate)
                    list.Add(peer);

                _peersHandlingAllMessages = list;
            }
            else
            {
                var list = _dynamicPeerSubscriptions
                    .Where(item => item.Peer.Id != peer.Id || !Equals(item.Subscription, subscription))
                    .ToList();

                if (isAddOrUpdate)
                    list.Add(new PeerSubscription(peer, subscription));

                Interlocked.Exchange(ref _dynamicPeerSubscriptions, list);
            }
        }

        /**
         * Get the results from browsing and put it into the sharedFileList folder of the peer.
         * Should maybe change the key to a Peer , but since implementation is unknown
         * no assurance that hashcode will be the same ...
         * */
        public void BrowsingTerminated(Peer peer, SearchResult Results, List<G2PacketQH2> resultPackets)
        {
            if (resultPackets.Count == 0)
            {
                RegroupResults(Results); // if no results from browsing directly sends results
                return;
            }
            Results.PeerCollection.Add(peer);
            Peer p = Results.PeerCollection.Find(peer);

            int fileCount = 0;
            // add results !!
            foreach(G2PacketQH2 qh2 in resultPackets)
            {
                foreach(G2Packet child in qh2.children)
                {
                    if (!child.type.Equals(G2PacketType.H))
                        continue;
                    G2PacketH hit = child as G2PacketH;
                    G2File file = G2File.ParseHit(hit,FileLocationFound.SharedLocalComputer);
                    if (file == null) continue;
                    p.SharedLocalfilesList.Add(file);
                    file.PeerDiffusedFiles.Add(p);
                    SharedTotalFiles++;
                    fileCount++;
                }
            }

            G2Log.Write("SearchResults : New Browsing Result from " + p.Ip + ":" + p.Port + " ==> " + p.Files.Count + " files && " + p.SharedLocalfilesList.Count + " shared files ...");

            RegroupResults(Results);
        }

 public DiscoverBuilder(Peer peer)
 {
     _peer = peer;
     PortUdp = Ports.DefaultPort;
     PortTcp = Ports.DefaultPort;
     DiscoverTimeoutSec = 5;
 }

        public void should_not_block_when_hitting_high_water_mark()
        {
            var senderTransport = CreateAndStartZmqTransport();
            senderTransport.SocketOptions.SendHighWaterMark = 3;
            senderTransport.SocketOptions.SendTimeout = 50.Milliseconds();
            senderTransport.SocketOptions.SendRetriesBeforeSwitchingToClosedState = 2;

            var receviedMessages = new List<TransportMessage>();
            var upReceiverTransport = CreateAndStartZmqTransport(onMessageReceived: receviedMessages.Add);
            var upReceiver = new Peer(new PeerId("Abc.Testing.Receiver.Up"), upReceiverTransport.InboundEndPoint);

            var downReceiverTransport = CreateAndStartZmqTransport();
            var downReceiver = new Peer(new PeerId("Abc.Testing.Receiver.Down"), downReceiverTransport.InboundEndPoint);

            downReceiverTransport.Stop();

            for (var i = 1; i <= 10; ++i)
            {
                var message = new FakeCommand(i).ToTransportMessage();
                senderTransport.Send(message, new[] { upReceiver, downReceiver });

                var expectedMessageCount = i;
                Wait.Until(() => receviedMessages.Count == expectedMessageCount, 500.Milliseconds(), "Failed to send message after " + i + " successful sent");
            }
        }

 public PeerEntry(PeerDescriptor descriptor, ConcurrentDictionary<MessageTypeId, PeerSubscriptionTree> globalSubscriptionsIndex)
 {
     Peer = new Peer(descriptor.Peer);
     IsPersistent = descriptor.IsPersistent;
     TimestampUtc = descriptor.TimestampUtc ?? DateTime.UtcNow;
     HasDebuggerAttached = descriptor.HasDebuggerAttached;
     
     _globalSubscriptionsIndex = globalSubscriptionsIndex;
 }

    void Start()
    {
        InputFieldAddress.text = "127.0.0.1";

        peer = new Peer(PORT);
        peer.onConnectionReceived += new Peer.ConnectionHandler(onPeerConnected);

        Debug.Log ("Socket alive: " + peer.socketAlive.ToString());
    }

        public void simple_subscription_key_should_simple_routing_key()
        {
            var peerSubscriptionTree = new PeerSubscriptionTree();
            var peer = new Peer(new PeerId("Abc.Testing.0"), "tcp://test:123");
            peerSubscriptionTree.Add(peer, new BindingKey("a"));

            var matchingPeer = peerSubscriptionTree.GetPeers(new BindingKey("a")).ExpectedSingle();
            matchingPeer.Id.ShouldEqual(peer.Id);
        }

        public static TransportMessage ToTransportMessage(this IMessage message, Peer sender = null)
        {
            sender = sender ?? new Peer(new PeerId("Abc.Testing.Peer"), "tcp://abctest:159");

            var serializer = new MessageSerializer();
            var messageBytes = serializer.Serialize(message);

            return new TransportMessage(message.TypeId(), messageBytes, sender);
        }

        public NewConnectionEventArgs(Peer peer, IConnection connection, TorrentManager manager)
            : base(manager)
        {
            if (!connection.IsIncoming && manager == null)
                throw new InvalidOperationException("An outgoing connection must specify the torrent manager it belongs to");

            this.connection = connection;
            this.peer = peer;
        }