public Watcher(TestLatch exitingLatch, TestLatch removedLatch, Address secondAddress)
            {
                _exitingLatch = exitingLatch;
                _removedLatch = removedLatch;
                _secondAddress = secondAddress;

                Receive<ClusterEvent.CurrentClusterState>(state =>
                {
                    if (state.Members.Any(m => m.Address == _secondAddress && m.Status == MemberStatus.Exiting))
                        _exitingLatch.CountDown();
                });
                Receive<ClusterEvent.MemberExited>(m =>
                {
                    if (m.Member.Address == secondAddress)
                    {
                        exitingLatch.CountDown();
                    }
                });
                Receive<ClusterEvent.MemberRemoved>(m =>
                {
                    if (m.Member.Address == secondAddress)
                    {
                        _removedLatch.CountDown();
                    }
                });
            }

        public void CreateCorrectToStringWithAddress()
        {
            var local = new Address("akka.tcp", "mysys");
            var a = new Address("akka.tcp", "mysys", "aaa", 2552);
            var b = new Address("akka.tcp", "mysys", "bb", 2552);
            var c = new Address("akka.tcp", "mysys", "cccc" , 2552);
            var root = new RootActorPath(local);
            root.ToStringWithAddress(a).ShouldBe("akka.tcp://[email protected]:2552/");
            (root / "user").ToStringWithAddress(a).ShouldBe("akka.tcp://[email protected]:2552/user");
            (root / "user" / "foo").ToStringWithAddress(a).ShouldBe("akka.tcp://[email protected]:2552/user/foo");

            root.ToStringWithAddress(b).ShouldBe("akka.tcp://[email protected]:2552/");
            (root / "user").ToStringWithAddress(b).ShouldBe("akka.tcp://[email protected]:2552/user");
            (root / "user" / "foo").ToStringWithAddress(b).ShouldBe("akka.tcp://[email protected]:2552/user/foo");

            root.ToStringWithAddress(c).ShouldBe("akka.tcp://[email protected]:2552/");
            (root / "user").ToStringWithAddress(c).ShouldBe("akka.tcp://[email protected]:2552/user");
            (root / "user" / "foo").ToStringWithAddress(c).ShouldBe("akka.tcp://[email protected]:2552/user/foo");

            var rootA = new RootActorPath(a);
            rootA.ToStringWithAddress(b).ShouldBe("akka.tcp://[email protected]:2552/");
            (rootA / "user").ToStringWithAddress(b).ShouldBe("akka.tcp://[email protected]:2552/user");
            (rootA / "user" / "foo").ToStringWithAddress(b).ShouldBe("akka.tcp://[email protected]:2552/user/foo");

        }

        private Address LocalAddressForRemote(
            IDictionary<string, HashSet<ProtocolTransportAddressPair>> transportMapping, Address remote)
        {
            HashSet<ProtocolTransportAddressPair> transports;

            if (transportMapping.TryGetValue(remote.Protocol, out transports))
            {
                ProtocolTransportAddressPair[] responsibleTransports =
                    transports.Where(t => t.ProtocolTransport.Transport.IsResponsibleFor(remote)).ToArray();
                if (responsibleTransports.Length == 0)
                {
                    throw new RemoteTransportException(
                        "No transport is responsible for address:[" + remote + "] although protocol [" + remote.Protocol +
                        "] is available." +
                        " Make sure at least one transport is configured to be responsible for the address.",
                        null);
                }
                if (responsibleTransports.Length == 1)
                {
                    return responsibleTransports.First().Address;
                }
                throw new RemoteTransportException(
                    "Multiple transports are available for [" + remote + ": " +
                    string.Join(",", responsibleTransports.Select(t => t.ToString())) + "] " +
                    "Remoting cannot decide which transport to use to reach the remote system. Change your configuration " +
                    "so that only one transport is responsible for the address.",
                    null);
            }

            throw new RemoteTransportException(
                "No transport is loaded for protocol: [" + remote.Protocol + "], available protocols: [" +
                string.Join(",", transportMapping.Keys.Select(t => t.ToString())) + "]", null);
        }

        /// <summary>
        /// Serializes the specified message.
        /// </summary>
        /// <param name="system">The system.</param>
        /// <param name="address"></param>
        /// <param name="message">The message.</param>
        /// <returns>SerializedMessage.</returns>
        public static SerializedMessage Serialize(ActorSystem system, Address address, object message)
        {
            Serializer serializer = system.Serialization.FindSerializerFor(message);

            SerializedMessage.Builder messageBuilder = new SerializedMessage.Builder()
                .SetMessage(ByteString.Unsafe.FromBytes(serializer.ToBinaryWithAddress(address, message)))
                .SetSerializerId(serializer.Identifier);

            var serializer2 = serializer as SerializerWithStringManifest;
            if (serializer2 != null)
            {
                var manifest = serializer2.Manifest(message);
                if (!string.IsNullOrEmpty(manifest))
                {
                    messageBuilder.SetMessageManifest(ByteString.CopyFromUtf8(manifest));
                }
            }
            else
            {
                if (serializer.IncludeManifest)
                    messageBuilder.SetMessageManifest(ByteString.CopyFromUtf8(message.GetType().AssemblyQualifiedName));
            }

            return messageBuilder.Build();
        }

        public RemoteMetricsSpec()
            : base(@"
            akka.actor.provider = ""Akka.Remote.RemoteActorRefProvider, Akka.Remote""
            akka.remote.log-frame-size-exceeding = 200 b
            akka.remote.helios.tcp = {
                port = 0
                hostname = localhost
            }
            akka.loglevel = DEBUG
            ")
        {
            _client = ActorSystem.Create("RemoteMetricsSpec-client", ConfigurationFactory.ParseString(@"
                akka.actor.provider =  ""Akka.Remote.RemoteActorRefProvider, Akka.Remote""
                 akka.remote.helios.tcp = {
                    port = 0
                    hostname = localhost
                }                
            ").WithFallback(Sys.Settings.Config));

            _address = Sys.AsInstanceOf<ExtendedActorSystem>().Provider.DefaultAddress;
            _subject = Sys.ActorOf(Props.Create(() => new Subject()).WithDeploy(Deploy.Local), "subject");
            var listener = Sys.ActorOf(Props.Create(() => new InfoEventListener(TestActor)).WithDeploy(Deploy.Local),
                "listener");
            Sys.EventStream.Subscribe(listener, typeof (Info));
        }

 public ConnectionAssociationHandle(IActorRef connection, Address localAddress, Address remoteAddress)
     : base(localAddress, remoteAddress)
 {
     _connection = connection;
     ReadHandlerSource = new TaskCompletionSource<IHandleEventListener>();
     ReadHandlerSource.Task.PipeTo(connection);
 }

        public RemoteDeploySpec()
            : base(@"
            akka {
                loglevel = INFO 
                log-dead-letters-during-shutdown = false
              //  actor.provider = ""Akka.Remote.RemoteActorRefProvider, Akka.Remote""
                remote.helios.tcp = {
                    hostname = localhost
                    port = 0
                }

                actor.deployment {
                  /router1 {
                    router = round-robin-pool
                    nr-of-instances = 3
                  }
                  /router2 {
                    router = round-robin-pool
                    nr-of-instances = 3
                  }
                  /router3 {
                    router = round-robin-pool
                    nr-of-instances = 0
                  }
                }
            }
")
        {
            _remoteSystem = ActorSystem.Create("RemoteSystem", Sys.Settings.Config);
            _remoteAddress = _remoteSystem.AsInstanceOf<ExtendedActorSystem>().Provider.DefaultAddress;
            var remoteAddressUid = AddressUidExtension.Uid(_remoteSystem);


        }

 private void Ready()
 {
     Receive<SendState>(msg =>
     {
         _cluster.SendCurrentClusterState(Self);
     });
     Receive<ClusterEvent.CurrentClusterState>(state =>
     {
         //_hub.ClusterState(state, _cluster.SelfAddress);
     });
     Receive<RemoveMe>(msg => {
         _logger.Warning("[l] Monitor stopping; Issuing a Cluster.Leave() command for following address: {0}", _cluster.SelfAddress);
         _cluster.Leave(_cluster.SelfAddress);
     });
     Receive<MemberLeave>(msg => {
         Address add = new Address(msg.Protocol, msg.System, msg.Host, msg.Port);
         _logger.Warning("[l] Forcing Member to leave cluster: {0}", add.ToString());
         _cluster.Leave(add);
     });
     Receive<MemberDown>(msg => {
         Address add = new Address(msg.Protocol, msg.System, msg.Host, msg.Port);
         _logger.Warning("[l] Forcing the Member down: {0}", add.ToString());
         _cluster.Down(add);
     });
 }

 public void RemoteActorRefProvider_should_create_RemoteActorRef_for_nonlocaladdress()
 {
     var nonLocalAddress = new Address("akka.trttl.gremlin.tcp", Sys.Name, "localhost", RARP.For(Sys).Provider.DefaultAddress.Port);
     var nonLocalActorPath = new RootActorPath(nonLocalAddress) / "user" / "foo";
     var resolved = RARP.For(Sys).Provider.ResolveActorRefWithLocalAddress(nonLocalActorPath.ToSerializationFormat(), nonLocalAddress);
     Assert.IsType<RemoteActorRef>(resolved); // should be a remote actorref
 }

 public DisassociatedEvent(Address localAddress, Address remoteAddress, bool inbound)
 {
     LocalAddress = localAddress;
     RemoteAddress = remoteAddress;
     IsInbound = inbound;
     EventName = "Disassociated";
 }

 public Message(IInternalActorRef recipient, Address recipientAddress, SerializedMessage serializedMessage, IActorRef senderOptional = null, SeqNo seq = null)
 {
     Seq = seq;
     SenderOptional = senderOptional;
     SerializedMessage = serializedMessage;
     RecipientAddress = recipientAddress;
     Recipient = recipient;
 }

 public void CreateCorrectToStringWithoutAddress()
 {
     var a = new Address("akka.tcp", "mysys");
     //TODO: there must be a / after system name
     new RootActorPath(a).ToStringWithoutAddress().ShouldBe("/");
     (new RootActorPath(a) / "user").ToStringWithoutAddress().ShouldBe("/user");
     (new RootActorPath(a) / "user" / "foo").ToStringWithoutAddress().ShouldBe("/user/foo");
     (new RootActorPath(a) / "user" / "foo" / "bar").ToStringWithoutAddress().ShouldBe("/user/foo/bar");
 }

 public IActorRef ReadOnlyEndpointFor(Address address)
 {
     IActorRef tmp;
     if (addressToReadonly.TryGetValue(address, out tmp))
     {
         return tmp;
     }
     return null;
 }

 public AssociationErrorEvent(Exception cause, Address localAddress, Address remoteAddress, bool inbound, LogLevel level)
 {
     LocalAddress = localAddress;
     RemoteAddress = remoteAddress;
     IsInbound = inbound;
     EventName = "AssociationError";
     _level = level;
     Cause = cause;
 }

 public void Setup(BenchmarkContext context)
 {
     _actorSystem = ActorSystem.Create("MessageDispatcher" + Counter.GetAndIncrement(), RemoteHocon);
     _systemAddress = RARP.For(_actorSystem).Provider.DefaultAddress;
     _inboundMessageDispatcherCounter = context.GetCounter(MessageDispatcherThroughputCounterName);
     _message = SerializedMessage.CreateBuilder().SetSerializerId(0).SetMessage(ByteString.CopyFromUtf8("foo")).Build();
     _dispatcher = new DefaultMessageDispatcher(_actorSystem, RARP.For(_actorSystem).Provider, _actorSystem.Log);
     _targetActorRef = new BenchmarkActorRef(_inboundMessageDispatcherCounter, RARP.For(_actorSystem).Provider);
 }

 /// <summary>
 /// Initializes a new instance of the <see cref="RemoteActorRef"/> class.
 /// </summary>
 /// <param name="remote">The remote.</param>
 /// <param name="localAddressToUse">The local address to use.</param>
 /// <param name="path">The path.</param>
 /// <param name="parent">The parent.</param>
 /// <param name="props">The props.</param>
 /// <param name="deploy">The deploy.</param>
 internal RemoteActorRef(RemoteTransport remote, Address localAddressToUse, ActorPath path, InternalActorRef parent,
     Props props, Deploy deploy)
 {
     Remote = remote;
     LocalAddressToUse = localAddressToUse;
     Path = path;
     _parent = parent;
     _props = props;
     _deploy = deploy;
 }

 public TcpTransport(ActorSystem system, Config config) : base(system, config)
 {
     string protocol = "akka." + config.GetString("transport-protocol");
     SchemeIdentifier = protocol;
     string host = config.GetString("hostname");
     int port = config.GetInt("port");
     Address = new Address(protocol, system.Name, host, port);
     log = Logging.GetLogger(system, this);
     server = new TcpServer(system, Address);
 }

 /// <summary>
 /// Converts a <see cref="ByteString"/> structure into a Helios <see cref="NetworkData"/> structure
 /// </summary>
 /// <param name="byteString">The data to send over the network</param>
 /// <param name="address">The address that we received data from / are sending data to</param>
 /// <returns>a new <see cref="NetworkData"/> struct</returns>
 public static NetworkData ToData(ByteString byteString, Address address)
 {
     var data = new NetworkData()
     {
         Buffer = byteString.ToByteArray(),
         RemoteHost = HeliosTransport.AddressToNode(address)
     };
     data.Length = data.Buffer.Length;
     return data;
 }

 public void MemoryFootprint(BenchmarkContext context)
 {
     var actorPaths = new Address[100000];
     for (var i = 0; i < 100000;)
     {
         actorPaths[i] = new Address("akka", "foo", "localhost", 9091);
         ++i;
         _parseThroughput.Increment();
     }
 }

 public override void Down(Address node)
 {
     if (_leader)
     {
         _probe.Tell(new DownCalled(node));
     }
     else
     {
         _probe.Tell("down must only be done by leader");
     }
 }