Set class to handle peer eventing service  disconnects

@param newPeerEventingServiceDisconnectedHandler New handler to replace.  If set to null no retry will take place.
@return the old handler.
507527
public PeerEventingServiceDisconnected setPeerEventingServiceDisconnected(PeerEventingServiceDisconnected newPeerEventingServiceDisconnectedHandler) {
        PeerEventingServiceDisconnected ret = disconnectedHandler;
        disconnectedHandler = newPeerEventingServiceDisconnectedHandler;
        return ret;
    }
569437
public ThriftConnectionHandle<T> recreateConnectionHandle() throws ThriftConnectionPoolException {
		ThriftConnectionHandle<T> handle = new ThriftConnectionHandle<>(this.thriftConnection,
				this.thriftConnectionPartition, this.thriftConnectionPool, true);
		handle.thriftConnectionPartition = this.thriftConnectionPartition;
		handle.connectionCreationTimeInMs = this.connectionCreationTimeInMs;
		handle.connectionLastResetInMs = this.connectionLastResetInMs;
		handle.connectionLastUsedInMs = this.connectionLastUsedInMs;
		handle.possiblyBroken = this.possiblyBroken;
		this.thriftConnection = null;

		return handle;
	}
730284
public boolean removeDisconnectedEventListener(PeerDisconnectedEventListener listener) {
        boolean result = ListenerRegistration.removeFromList(listener, peerDisconnectedEventListeners);
        for (Peer peer : getConnectedPeers())
            peer.removeDisconnectedEventListener(listener);
        for (Peer peer : getPendingPeers())
            peer.removeDisconnectedEventListener(listener);
        return result;
    }
224290
private void add(Peer peer) {
		String id = peer.getId();
		peers.put(id, peer);
		timeouts.put(id, executor.schedule(new Deactivate(id), delay, MILLISECONDS));
		notifyConnected(new Builder(peer).status(Status.CONNECTED).build());
		notifyActive(peer);
	}
358563
protected void firePeerDisconnected(URI peerURI, SpaceID space) {
        final NetworkServiceListener[] ilisteners;
        synchronized (this.listeners) {
            ilisteners = new NetworkServiceListener[this.listeners.size()];
            this.listeners.toArray(ilisteners);
        }
        for (final NetworkServiceListener listener : ilisteners) {
            listener.peerDisconnected(peerURI, space);
        }
    }
260474
private void handleDisconnectEvent(DisconnectEvent disconnectEvent)
    {
        // reset version information as it might have changed while Asterisk
        // restarted
        version = null;
        versions = null;

        // same for channels, agents and queues rooms, they are reinitialized
        // when reconnected
        channelManager.disconnected();
        agentManager.disconnected();
        meetMeManager.disconnected();
        queueManager.disconnected();
        initialized = false;
        initializing = false;
    }
396872
@Override
  public <T> EventHandler<T> getHandler(
      final RemoteIdentifier destinationIdentifier, final Class<? extends T> messageType) {

    if (LOG.isLoggable(Level.FINE)) {
      LOG.log(Level.FINE, "RemoteManager: {0} destinationIdentifier: {1} messageType: {2}",
          new Object[] {this.name, destinationIdentifier, messageType.getCanonicalName()});
    }

    return new ProxyEventHandler<>(this.myIdentifier, destinationIdentifier,
        "default", this.reSendStage.<T>getHandler(), this.seqGen);
  }
148430
public void setNewInstanceResolver(final Object newInstanceResolver) {
        if (newInstanceResolver instanceof NewInstanceResolver) {
            this.newInstanceResolver = (NewInstanceResolver) newInstanceResolver;
        } else if (newInstanceResolver instanceof Closure) {
            final ObjectGraphBuilder self = this;
            this.newInstanceResolver = new NewInstanceResolver() {
                public Object newInstance(Class klass, Map attributes)
                        throws InstantiationException, IllegalAccessException {
                    Closure cls = (Closure) newInstanceResolver;
                    cls.setDelegate(self);
                    return cls.call(klass, attributes);
                }
            };
        } else {
            this.newInstanceResolver = new DefaultNewInstanceResolver();
        }
    }
1027920
@Override
  public void handlePeerReady(SharingPeer peer) {
    initIfNecessary(peer);

    RequestsCollection requestsCollection = getRequestsCollection(peer);
    requestsCollection.sendAllRequests();
  }
567599
@ObjectiveCName("changeNotificationsEnabledWithPeer:withValue:")
    public void changeNotificationsEnabled(Peer peer, boolean val) {
        modules.getSettingsModule().changeNotificationsEnabled(peer, val);
    }
392582
@Override
  public void _Network_Peers_removePeer(String name) throws MBeanException {
    try {
      MutablePeerTable n = (MutablePeerTable) stack.unwrap(PeerTable.class);
      n.removePeer(name);
    }
    catch (InternalException e) {
      throw new MBeanException(e, "Failed to remove peer with name '" + name + "'");
    }
  }
487561
@Override
    public void afterPropertiesSet() throws Exception {
        // create the nio acceptor
        acceptor = new NioSocketAcceptor(Runtime.getRuntime().availableProcessors() * 4);
        acceptor.addListener(new IoServiceListener() {

            @Override
            public void serviceActivated(IoService service) throws Exception {
                //log.debug("serviceActivated: {}", service);
            }

            @Override
            public void serviceIdle(IoService service, IdleStatus idleStatus) throws Exception {
                //logger.debug("serviceIdle: {} status: {}", service, idleStatus);
            }

            @Override
            public void serviceDeactivated(IoService service) throws Exception {
                //log.debug("serviceDeactivated: {}", service);
            }

            @Override
            public void sessionCreated(IoSession session) throws Exception {
                log.debug("sessionCreated: {}", session);
                //log.trace("Acceptor sessions: {}", acceptor.getManagedSessions());
            }

            @Override
            public void sessionClosed(IoSession session) throws Exception {
                log.debug("sessionClosed: {}", session);
            }

            @Override
            public void sessionDestroyed(IoSession session) throws Exception {
                //log.debug("sessionDestroyed: {}", session);
            }

        });
        // configure the acceptor
        SocketSessionConfig sessionConf = acceptor.getSessionConfig();
        sessionConf.setReuseAddress(true);
        sessionConf.setTcpNoDelay(true);
        sessionConf.setSendBufferSize(sendBufferSize);
        sessionConf.setReadBufferSize(receiveBufferSize);
        // prevent the background blocking queue
        sessionConf.setUseReadOperation(false);
        // seconds
        sessionConf.setWriteTimeout(writeTimeout);
        // set an idle time in seconds
        if (idleTimeout > 0) {
            sessionConf.setIdleTime(IdleStatus.BOTH_IDLE, idleTimeout);
        }
        // close sessions when the acceptor is stopped
        acceptor.setCloseOnDeactivation(true);
        // requested maximum length of the queue of incoming connections
        acceptor.setBacklog(64);
        acceptor.setReuseAddress(true);
        // instance the websocket handler
        if (ioHandler == null) {
            ioHandler = new WebSocketHandler();
        }
        log.trace("I/O handler: {}", ioHandler);
        acceptor.setHandler(ioHandler);
        DefaultIoFilterChainBuilder chain = acceptor.getFilterChain();
        // if handling wss init the config
        SslFilter sslFilter = null;
        if (secureConfig != null) {
            try {
                sslFilter = secureConfig.getSslFilter();
                chain.addFirst("sslFilter", sslFilter);
            } catch (Exception e) {
                log.warn("SSL configuration failed, websocket will not be secure", e);
            }
        }
        if (log.isTraceEnabled()) {
            chain.addLast("logger", new LoggingFilter());
        }
        // add the websocket codec factory
        chain.addLast("protocol", new ProtocolCodecFilter(new WebSocketCodecFactory()));
        if (addresses.isEmpty()) {
            if (sslFilter != null) {
                log.info("WebSocket (wss) will be bound to port {}", port);
            } else {
                log.info("WebSocket (ws) will be bound to port {}", port);
            }
            acceptor.bind(new InetSocketAddress(port));
        } else {
            if (sslFilter != null) {
                log.info("WebSocket (wss) will be bound to {}", addresses);
            } else {
                log.info("WebSocket (ws) will be bound to {}", addresses);
            }
            try {
                // loop through the addresses and bind
                Set<InetSocketAddress> socketAddresses = new HashSet<InetSocketAddress>();
                for (String addr : addresses) {
                    if (addr.indexOf(':') != -1) {
                        String[] parts = addr.split(":");
                        socketAddresses.add(new InetSocketAddress(parts[0], Integer.valueOf(parts[1])));
                    } else {
                        socketAddresses.add(new InetSocketAddress(addr, port));
                    }
                }
                log.debug("Binding to {}", socketAddresses.toString());
                acceptor.bind(socketAddresses);
            } catch (Exception e) {
                log.warn("Exception occurred during resolve / bind", e);
            }
        }
        log.info("Started {} websocket transport. Timeouts - idle: {} write: {}", (isSecure() ? "secure" : ""), idleTimeout, writeTimeout);
        if (log.isDebugEnabled()) {
            log.debug("Acceptor sizes - send: {} recv: {}", acceptor.getSessionConfig().getSendBufferSize(), acceptor.getSessionConfig().getReadBufferSize());
        }
    }
619076
@Deprecated
   public Class<? extends org.infinispan.client.hotrod.impl.transport.tcp.FailoverRequestBalancingStrategy> balancingStrategyClass() {
      FailoverRequestBalancingStrategy strategy = balancingStrategyFactory.get();
      if (org.infinispan.client.hotrod.impl.transport.tcp.FailoverRequestBalancingStrategy.class.isInstance(strategy)) {
         return (Class<? extends org.infinispan.client.hotrod.impl.transport.tcp.FailoverRequestBalancingStrategy>) strategy.getClass();
      } else {
         return org.infinispan.client.hotrod.impl.transport.tcp.FailoverRequestBalancingStrategy.class;
      }
   }
782078
private void update(Peer peer) {
		String id = peer.getId();
		Peer previous = peers.get(id);
		switch(previous.getStatus()) {
			case ACTIVE:
				timeouts.get(id).cancel(true);
				timeouts.put(id, executor.schedule(new Deactivate(id), delay, MILLISECONDS));
				break;
			case INACTIVE:
				peers.put(peer.getId(), peer);
				timeouts.get(id).cancel(true);
				timeouts.put(id, executor.schedule(new Deactivate(id), delay, MILLISECONDS));
				notifyActive(peer);
				break;
			default:
				String error = "Unexpected peer status: " + previous.getStatus();
				throw new RuntimeException(error);
		}
	}
1257022
void connectEnvelope(Envelope envelope) throws TransactionException {

        if (shutdown) {
            logger.warn(format("%s not connecting is shutdown.", toString()));
            return;
        }

        final AtomicBoolean retry = new AtomicBoolean(true); // make sure we only retry connection once for each connection attempt.

        ManagedChannel lmanagedChannel = managedChannel;

        if (lmanagedChannel == null || lmanagedChannel.isTerminated() || lmanagedChannel.isShutdown()) {

            lmanagedChannel = channelBuilder.build();
            managedChannel = lmanagedChannel;

        }

        try {

            DeliverGrpc.DeliverStub broadcast = DeliverGrpc.newStub(lmanagedChannel);

            // final DeliverResponse[] ret = new DeliverResponse[1];
            //   final List<DeliverResponse> retList = new ArrayList<>();
            final List<Throwable> throwableList = new ArrayList<>();
            final CountDownLatch finishLatch = new CountDownLatch(1);

            so = new StreamObserver<DeliverResponse>() {

                @Override
                public void onNext(DeliverResponse resp) {

                    // logger.info("Got Broadcast response: " + resp);
                    logger.trace(format("DeliverResponse %s resp status value:%d  status %s, typecase %s ",
                            PeerEventServiceClient.this.toString(), resp.getStatusValue(), resp.getStatus(), resp.getTypeCase()));

                    final DeliverResponse.TypeCase typeCase = resp.getTypeCase();

                    if (typeCase == STATUS) {

                        logger.debug(format("DeliverResponse  %s setting done.",
                                PeerEventServiceClient.this.toString()));

                        if (resp.getStatus() == Common.Status.SUCCESS) { // unlike you may think this only happens when all blocks are fetched.
                            peer.setLastConnectTime(System.currentTimeMillis());
                            peer.resetReconnectCount();
                        } else {

                            final long rec = peer.getReconnectCount();

                            PeerEventingServiceException peerEventingServiceException = new PeerEventingServiceException(format("%s attempts %s Status returned failure code %d (%s) during peer service event registration",
                                    PeerEventServiceClient.this.toString(), rec, resp.getStatusValue(), resp.getStatus().name()));
                            peerEventingServiceException.setResponse(resp);
                            if (rec % 10 == 0) {
                                logger.warn(PeerEventServiceClient.this.toString() + " " + peerEventingServiceException.getMessage());
                            }

                            throwableList.add(peerEventingServiceException);
                        }

                    } else if (typeCase == FILTERED_BLOCK || typeCase == BLOCK) {
                        if (typeCase == BLOCK) {
                            logger.trace(format("%s got event block hex hashcode: %016x, block number: %d",
                                    PeerEventServiceClient.this.toString(), resp.getBlock().hashCode(), resp.getBlock().getHeader().getNumber()));
                        } else {
                            logger.trace(format("%s got event block hex hashcode: %016x, block number: %d",
                                    PeerEventServiceClient.this.toString(), resp.getFilteredBlock().hashCode(), resp.getFilteredBlock().getNumber()));
                        }

                        peer.setLastConnectTime(System.currentTimeMillis());
                        long reconnectCount = peer.getReconnectCount();
                        if (reconnectCount > 1) {
                            logger.info(format("%s reconnected after %d attempts on channel %s, peer %s, url %s",
                                    PeerEventServiceClient.this.toString(), reconnectCount, channelName, name, url));
                        }
                        peer.resetReconnectCount();

                        BlockEvent blockEvent = new BlockEvent(peer, resp);
                        peer.setLastBlockSeen(blockEvent);

                        channelEventQue.addBEvent(blockEvent);
                    } else {
                        logger.error(format("%s got event block with unknown type: %s, %d",
                                PeerEventServiceClient.this.toString(), typeCase.name(), typeCase.getNumber()));

                        PeerEventingServiceException peerEventingServiceException = new PeerEventingServiceException(format("% got event block with unknown type: %s, %d",
                                PeerEventServiceClient.this.toString(), typeCase.name(), typeCase.getNumber()));
                        peerEventingServiceException.setResponse(resp);

                        throwableList.add(peerEventingServiceException);

                    }
                    finishLatch.countDown();

                }

                @Override
                public void onError(Throwable t) {
                    ManagedChannel llmanagedChannel = managedChannel;
                    if (llmanagedChannel != null) {
                        try {
                            llmanagedChannel.shutdownNow();
                        } catch (Exception e) {
                            logger.warn(format("Received error on %s, attempts %d. %s shut down of grpc channel.",
                                    PeerEventServiceClient.this.toString(), peer == null ? -1 : peer.getReconnectCount(), e.getMessage()), e);
                        }
                        managedChannel = null;
                    }
                    if (!shutdown) {
                        final long reconnectCount = peer.getReconnectCount();
                        if (PEER_EVENT_RECONNECTION_WARNING_RATE > 1 && reconnectCount % PEER_EVENT_RECONNECTION_WARNING_RATE == 1) {
                            logger.warn(format("Received error on  %s, attempts %d. %s",
                                    PeerEventServiceClient.this.toString(), reconnectCount, t.getMessage()));

                        } else {
                            logger.trace(format("Received error on %s, attempts %d. %s",
                                    PeerEventServiceClient.this.toString(), reconnectCount, t.getMessage()));

                        }

                        if (retry.getAndSet(false)) {
                            peer.reconnectPeerEventServiceClient(PeerEventServiceClient.this, t);
                        }

                    }
                    finishLatch.countDown();
                }

                @Override
                public void onCompleted() {
                    logger.debug(format("DeliverResponse onCompleted %s setting done.",
                            PeerEventServiceClient.this.toString()));
                    //            done = true;
                    //There should have been a done before this...
                    finishLatch.countDown();
                }
            };

            nso = filterBlock ? broadcast.deliverFiltered(so) : broadcast.deliver(so);

            nso.onNext(envelope);

            // try {
            if (!finishLatch.await(peerEventRegistrationWaitTimeMilliSecs, TimeUnit.MILLISECONDS)) {
                PeerEventingServiceException ex = new PeerEventingServiceException(format(
                        "Channel %s connect time exceeded for peer eventing service %s, timed out at %d ms.", channelName, name, peerEventRegistrationWaitTimeMilliSecs));
                ex.setTimedOut(peerEventRegistrationWaitTimeMilliSecs);
                logger.warn(toString() + " " + ex.getMessage());
                throwableList.add(0, ex);

            }
            logger.trace(toString() + " done waiting for reply!");

            if (!throwableList.isEmpty()) {
                ManagedChannel llmanagedChannel = managedChannel;
                if (llmanagedChannel != null) {
                    llmanagedChannel.shutdownNow();
                    managedChannel = null;
                }
                Throwable throwable = throwableList.get(0);
                if (retry.getAndSet(false)) {
                    peer.reconnectPeerEventServiceClient(this, throwable);
                }

            }

        } catch (InterruptedException e) {
            ManagedChannel llmanagedChannel = managedChannel;
            if (llmanagedChannel != null) {
                llmanagedChannel.shutdownNow();
                managedChannel = null;
            }
            logger.error(toString() + " error message: " + e.getMessage(), e); // not likely

            if (retry.getAndSet(false)) {
                peer.reconnectPeerEventServiceClient(this, e);
            }

        } finally {
            if (null != nso) {

                try {
                    nso.onCompleted();
                } catch (Exception e) {  //Best effort only report on debug
                    logger.debug(format("Exception completing connect with %s %s",
                            toString(), e.getMessage()), e);
                }

            }
        }
    }
613560
protected void refreshPeer (NodeRecord record)
    {
        PeerNode peer = _peers.get(record.nodeName);
        if (peer == null) {
            peer = _injector.getInstance(getPeerNodeClass());
            _peers.put(record.nodeName, peer);
            peer.init(record);
        }
        peer.refresh(record);
    }
921150
public void addConnectedEventListener(Executor executor, PeerConnectedEventListener listener) {
        peerConnectedEventListeners.add(new ListenerRegistration<>(checkNotNull(listener), executor));
        for (Peer peer : getConnectedPeers())
            peer.addConnectedEventListener(executor, listener);
        for (Peer peer : getPendingPeers())
            peer.addConnectedEventListener(executor, listener);
    }