public void Equals_should_return_expected_result(
            int port1,
            int localValue1,
            int serverValue1,
            int port2,
            int localValue2,
            int serverValue2,
            bool expectedEqualsResult,
            bool expectedStructurallyEqualsResult)
        {
            var clusterId = new ClusterId();
            var serverId1 = new ServerId(clusterId, new DnsEndPoint("localhost", port1));
            var serverId2 = new ServerId(clusterId, new DnsEndPoint("localhost", port2));

            var subject1 = new ConnectionId(serverId1, localValue1).WithServerValue(serverValue1);
            var subject2 = new ConnectionId(serverId2, localValue2).WithServerValue(serverValue2);

            // note: Equals ignores the server values and StructurallyEquals compares all fields
            var equalsResult1 = subject1.Equals(subject2);
            var equalsResult2 = subject2.Equals(subject1);
            var structurallyEqualsResult1 = subject1.StructurallyEquals(subject2);
            var structurallyEqualsResult2 = subject2.StructurallyEquals(subject1);
            var hashCode1 = subject1.GetHashCode();
            var hashCode2 = subject2.GetHashCode();

            equalsResult1.Should().Be(expectedEqualsResult);
            equalsResult2.Should().Be(expectedEqualsResult);
            structurallyEqualsResult1.Should().Be(expectedStructurallyEqualsResult);
            structurallyEqualsResult2.Should().Be(expectedStructurallyEqualsResult);
            (hashCode1 == hashCode2).Should().Be(expectedEqualsResult);
        }

        public void LocalValues_of_2_ids_should_not_be_the_same_when_automically_constructed()
        {
            var subject = new ConnectionId(__serverId);
            var subject2 = new ConnectionId(__serverId);

            subject.LocalValue.Should().NotBe(subject2.LocalValue);
        }

 /// <summary>
 /// Initializes a new instance of the <see cref="ConnectionReceivingMessageFailedEvent" /> struct.
 /// </summary>
 /// <param name="connectionId">The connection identifier.</param>
 /// <param name="responseTo">The id of the message we were receiving a response to.</param>
 /// <param name="exception">The exception.</param>
 /// <param name="operationId">The operation identifier.</param>
 public ConnectionReceivingMessageFailedEvent(ConnectionId connectionId, int responseTo, Exception exception, long? operationId)
 {
     _connectionId = connectionId;
     _responseTo = responseTo;
     _exception = exception;
     _operationId = operationId;
 }

 /// <summary>
 /// Initializes a new instance of the <see cref="ConnectionOpeningFailedEvent" /> struct.
 /// </summary>
 /// <param name="connectionId">The connection identifier.</param>
 /// <param name="connectionSettings">The connection settings.</param>
 /// <param name="exception">The exception.</param>
 /// <param name="operationId">The operation identifier.</param>
 public ConnectionOpeningFailedEvent(ConnectionId connectionId, ConnectionSettings connectionSettings, Exception exception, long? operationId)
 {
     _connectionId = connectionId;
     _connectionSettings = connectionSettings;
     _exception = exception;
     _operationId = operationId;
 }

 /// <summary>
 /// Initializes a new instance of the <see cref="ConnectionOpenedEvent" /> struct.
 /// </summary>
 /// <param name="connectionId">The connection identifier.</param>
 /// <param name="connectionSettings">The connection settings.</param>
 /// <param name="duration">The duration of time it took to open the connection.</param>
 /// <param name="operationId">The operation identifier.</param>
 public ConnectionOpenedEvent(ConnectionId connectionId, ConnectionSettings connectionSettings, TimeSpan duration, long? operationId)
 {
     _connectionId = connectionId;
     _connectionSettings = connectionSettings;
     _duration = duration;
     _operationId = operationId;
 }

        /// <summary>
        /// Maps the specified response to a custom exception (if possible).
        /// </summary>
        /// <param name="connectionId">The connection identifier.</param>
        /// <param name="response">The response.</param>
        /// <returns>
        /// The custom exception (or null if the response could not be mapped to a custom exception).
        /// </returns>
        public static Exception Map(ConnectionId connectionId, BsonDocument response)
        {
            BsonValue code;
            if (response.TryGetValue("code", out code) && code.IsNumeric)
            {
                switch (code.ToInt32())
                {
                    case 50:
                    case 13475:
                    case 16986:
                    case 16712:
                        return new MongoExecutionTimeoutException(connectionId, message: "Operation exceeded time limit.");
                }
            }

            // the server sometimes sends a response that is missing the "code" field but does have an "errmsg" field
            BsonValue errmsg;
            if (response.TryGetValue("errmsg", out errmsg) && errmsg.IsString)
            {
                if (errmsg.AsString.Contains("exceeded time limit") ||
                    errmsg.AsString.Contains("execution terminated"))
                {
                    return new MongoExecutionTimeoutException(connectionId, message: "Operation exceeded time limit.");
                }
            }

            return null;
        }

 /// <summary>
 /// Initializes a new instance of the <see cref="ConnectionSendingMessagesFailedEvent" /> struct.
 /// </summary>
 /// <param name="connectionId">The connection identifier.</param>
 /// <param name="requestIds">The request ids.</param>
 /// <param name="exception">The exception.</param>
 /// <param name="operationId">The operation identifier.</param>
 public ConnectionSendingMessagesFailedEvent(ConnectionId connectionId, IReadOnlyList<int> requestIds, Exception exception, long? operationId)
 {
     _connectionId = connectionId;
     _requestIds = requestIds;
     _exception = exception;
     _operationId = operationId;
 }

 public void TestFixtureSetup()
 {
     _connectionId = new ConnectionId(new ServerId(new ClusterId(1), new DnsEndPoint("localhost", 27017)), 2);
     _innerException = new Exception("inner");
     _writeConcernError = new WriteConcernError(1, "writeConcernError", new BsonDocument("details", "writeConcernError"));
     _writeError = new WriteError(ServerErrorCategory.Uncategorized, 1, "writeError", new BsonDocument("details", "writeError"));
 }

        // constructors
        public BinaryConnection(ServerId serverId, EndPoint endPoint, ConnectionSettings settings, IStreamFactory streamFactory, IConnectionInitializer connectionInitializer, IEventSubscriber eventSubscriber)
        {
            Ensure.IsNotNull(serverId, nameof(serverId));
            _endPoint = Ensure.IsNotNull(endPoint, nameof(endPoint));
            _settings = Ensure.IsNotNull(settings, nameof(settings));
            _streamFactory = Ensure.IsNotNull(streamFactory, nameof(streamFactory));
            _connectionInitializer = Ensure.IsNotNull(connectionInitializer, nameof(connectionInitializer));
            Ensure.IsNotNull(eventSubscriber, nameof(eventSubscriber));

            _backgroundTaskCancellationTokenSource = new CancellationTokenSource();
            _backgroundTaskCancellationToken = _backgroundTaskCancellationTokenSource.Token;

            _connectionId = new ConnectionId(serverId);
            _receiveCoordinator = new ReceiveCoordinator();
            _sendLock = new SemaphoreSlim(1);
            _state = new InterlockedInt32(State.Initial);

            _commandEventHelper = new CommandEventHelper(eventSubscriber);
            eventSubscriber.TryGetEventHandler(out _failedEventHandler);
            eventSubscriber.TryGetEventHandler(out _closingEventHandler);
            eventSubscriber.TryGetEventHandler(out _closedEventHandler);
            eventSubscriber.TryGetEventHandler(out _openingEventHandler);
            eventSubscriber.TryGetEventHandler(out _openedEventHandler);
            eventSubscriber.TryGetEventHandler(out _failedOpeningEventHandler);
            eventSubscriber.TryGetEventHandler(out _receivingMessageEventHandler);
            eventSubscriber.TryGetEventHandler(out _receivedMessageEventHandler);
            eventSubscriber.TryGetEventHandler(out _failedReceivingMessageEventHandler);
            eventSubscriber.TryGetEventHandler(out _sendingMessagesEventHandler);
            eventSubscriber.TryGetEventHandler(out _sentMessagesEventHandler);
            eventSubscriber.TryGetEventHandler(out _failedSendingMessagesEvent);
        }

 /// <summary>
 /// Initializes a new instance of the <see cref="ConnectionReceivedMessageEvent" /> struct.
 /// </summary>
 /// <param name="connectionId">The connection identifier.</param>
 /// <param name="responseTo">The id of the message we received a response to.</param>
 /// <param name="length">The length of the received message.</param>
 /// <param name="networkDuration">The duration of network time it took to receive the message.</param>
 /// <param name="deserializationDuration">The duration of deserialization time it took to receive the message.</param>
 public ConnectionReceivedMessageEvent(ConnectionId connectionId, int responseTo, int length, TimeSpan networkDuration, TimeSpan deserializationDuration)
 {
     _connectionId = connectionId;
     _responseTo = responseTo;
     _length = length;
     _networkDuration = networkDuration;
     _deserializationDuration = deserializationDuration;
 }

 public static string Format(ConnectionId id)
 {
     if (id.ServerValue.HasValue)
     {
         return id.LocalValue.ToString() + "-" + id.ServerValue.Value.ToString();
     }
     return id.LocalValue.ToString();
 }

 /// <summary>
 /// Initializes a new instance of the <see cref="ConnectionSentMessagesEvent" /> struct.
 /// </summary>
 /// <param name="connectionId">The connection identifier.</param>
 /// <param name="requestIds">The request ids.</param>
 /// <param name="length">The length.</param>
 /// <param name="networkDuration">The duration of time spent on the network.</param>
 /// <param name="serializationDuration">The duration of time spent serializing the messages.</param>
 public ConnectionSentMessagesEvent(ConnectionId connectionId, IReadOnlyList<int> requestIds, int length, TimeSpan networkDuration, TimeSpan serializationDuration)
 {
     _connectionId = connectionId;
     _requestIds = requestIds;
     _length = length;
     _networkDuration = networkDuration;
     _serializationDuration = serializationDuration;
 }

 // static methods
 private static string FormatMessage(ConnectionId connectionId, long cursorId)
 {
     return string.Format(
         "Cursor {0} not found on server {1} using connection {2}.",
         cursorId,
         EndPointHelper.ToString(connectionId.ServerId.EndPoint),
         connectionId.ServerValue);
 }

 /// <summary>
 /// Initializes a new instance of the <see cref="CommandFailedEvent" /> struct.
 /// </summary>
 /// <param name="commandName">Name of the command.</param>
 /// <param name="exception">The exception.</param>
 /// <param name="operationId">The operation identifier.</param>
 /// <param name="requestId">The request identifier.</param>
 /// <param name="connectionId">The connection identifier.</param>
 /// <param name="duration">The duration.</param>
 public CommandFailedEvent(string commandName, Exception exception, long? operationId, int requestId, ConnectionId connectionId, TimeSpan duration)
 {
     _commandName = Ensure.IsNotNullOrEmpty(commandName, "commandName");
     _exception = Ensure.IsNotNull(exception, "exception");
     _connectionId = Ensure.IsNotNull(connectionId, "connectionId");
     _operationId = operationId;
     _requestId = requestId;
     _duration = duration;
 }

 /// <summary>
 /// Initializes a new instance of the <see cref="CommandSucceededEvent"/> struct.
 /// </summary>
 /// <param name="commandName">Name of the command.</param>
 /// <param name="reply">The reply.</param>
 /// <param name="operationId">The operation identifier.</param>
 /// <param name="requestId">The request identifier.</param>
 /// <param name="connectionId">The connection identifier.</param>
 /// <param name="duration">The duration.</param>
 public CommandSucceededEvent(string commandName, BsonDocument reply, long? operationId, int requestId, ConnectionId connectionId, TimeSpan duration)
 {
     _commandName = Ensure.IsNotNullOrEmpty(commandName, "commandName");
     _reply = Ensure.IsNotNull(reply, "reply");
     _connectionId = Ensure.IsNotNull(connectionId, "connectionId");
     _operationId = operationId;
     _requestId = requestId;
     _duration = duration;
 }

 /// <summary>
 /// Initializes a new instance of the <see cref="CommandStartedEvent" /> class.
 /// </summary>
 /// <param name="commandName">Name of the command.</param>
 /// <param name="command">The command.</param>
 /// <param name="databaseNamespace">The database namespace.</param>
 /// <param name="operationId">The operation identifier.</param>
 /// <param name="requestId">The request identifier.</param>
 /// <param name="connectionId">The connection identifier.</param>
 public CommandStartedEvent(string commandName, BsonDocument command, DatabaseNamespace databaseNamespace, long? operationId, int requestId, ConnectionId connectionId)
 {
     _commandName = Ensure.IsNotNullOrEmpty(commandName, "commandName");
     _command = Ensure.IsNotNull(command, "command");
     _databaseNamespace = Ensure.IsNotNull(databaseNamespace, "databaseNamespace");
     _connectionId = Ensure.IsNotNull(connectionId, "connectionId");
     _operationId = operationId;
     _requestId = requestId;
 }

        public virtual bool Equals(ConnectionId that)
        {
            if(!Equals(this.Value, that.Value))
            {
                return false;
            }

            return true;
        }

 internal TcpConnection(ConnectionId connectionId, uint sequenceNumber, Action<TcpConnection>removeCallback, string snifferType)
 {
     ConnectionId = connectionId;
     Source = connectionId.Source;
     Destination = connectionId.Destination;
     InitialSequenceNumber = sequenceNumber;
     RemoveCallback = () => removeCallback(this);
     SnifferType = snifferType;
 }

        private void Receive(IpPacket ipPacket)
        {
            var protocol = ipPacket.Protocol;
            if (protocol != IPProtocolType.TCP)
            {
                return;
            }
            if (ipPacket.PayloadPacket == null)
            {
                return;
            }
            var tcpPacket = new TcpPacket(new ByteArraySegment(ipPacket.PayloadPacket.BytesHighPerformance));

            var isFirstPacket = tcpPacket.Syn;
            var sourceIp = new IPAddress(ipPacket.SourceAddress.GetAddressBytes()).ToString();
            var destinationIp = new IPAddress(ipPacket.DestinationAddress.GetAddressBytes()).ToString();
            var connectionId = new ConnectionId(sourceIp, tcpPacket.SourcePort, destinationIp, tcpPacket.DestinationPort);

            lock (_lock)
            {
                TcpConnection connection;
                bool isInterestingConnection;
                if (isFirstPacket)
                {
                    connection = new TcpConnection(connectionId, tcpPacket.SequenceNumber);
                    OnNewConnection(connection);
                    isInterestingConnection = connection.HasSubscribers;
                    if (!isInterestingConnection)
                    {
                        return;
                    }
                    _connections[connectionId] = connection;
                    Debug.Assert(ipPacket.PayloadPacket.PayloadData.Length == 0);
                }
                else
                {
                    isInterestingConnection = _connections.TryGetValue(connectionId, out connection);
                    if (!isInterestingConnection)
                    {
                        return;
                    }

                    if (!string.IsNullOrEmpty(TcpLogFile))
                    {
                        File.AppendAllText(TcpLogFile,
                            string.Format("{0} {1}+{4} | {2} {3}+{4} ACK {5} ({6})\r\n",
                                connection.CurrentSequenceNumber, tcpPacket.SequenceNumber, connection.BytesReceived,
                                connection.SequenceNumberToBytesReceived(tcpPacket.SequenceNumber),
                                ipPacket.PayloadLength, tcpPacket.AcknowledgmentNumber,
                                connection.BufferedPacketDescription));
                    }
                    connection.HandleTcpReceived(tcpPacket.SequenceNumber,
                        new ByteArraySegment(ipPacket.PayloadPacket.PayloadData));
                }
            }
        }

 /// <summary>
 /// Initializes a new instance of the <see cref="MongoWriteException" /> class.
 /// </summary>
 /// <param name="connectionId">The connection identifier.</param>
 /// <param name="writeError">The write error.</param>
 /// <param name="writeConcernError">The write concern error.</param>
 /// <param name="innerException">The inner exception.</param>
 public MongoWriteException(
     ConnectionId connectionId,
     WriteError writeError,
     WriteConcernError writeConcernError,
     Exception innerException)
     : base(connectionId, FormatMessage(writeError, writeConcernError), innerException)
 {
     _writeError = writeError;
     _writeConcernError = writeConcernError;
 }