/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.distributed.internal.membership.gms.messenger;

import static org.apache.geode.distributed.internal.membership.gms.GMSUtil.replaceStrings;
import static org.apache.geode.distributed.internal.membership.gms.messages.AbstractGMSMessage.ALL_RECIPIENTS;
import static org.apache.geode.internal.serialization.DataSerializableFixedID.FIND_COORDINATOR_REQ;
import static org.apache.geode.internal.serialization.DataSerializableFixedID.FIND_COORDINATOR_RESP;
import static org.apache.geode.internal.serialization.DataSerializableFixedID.JOIN_REQUEST;
import static org.apache.geode.internal.serialization.DataSerializableFixedID.JOIN_RESPONSE;

import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.UnknownHostException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap;
import org.apache.logging.log4j.Logger;
import org.jgroups.Address;
import org.jgroups.Channel;
import org.jgroups.Event;
import org.jgroups.JChannel;
import org.jgroups.Message.Flag;
import org.jgroups.Message.TransientFlag;
import org.jgroups.Receiver;
import org.jgroups.ReceiverAdapter;
import org.jgroups.View;
import org.jgroups.ViewId;
import org.jgroups.conf.ClassConfigurator;
import org.jgroups.protocols.UDP;
import org.jgroups.protocols.pbcast.NAKACK2;
import org.jgroups.protocols.pbcast.NakAckHeader2;
import org.jgroups.stack.IpAddress;
import org.jgroups.util.Digest;
import org.jgroups.util.UUID;

import org.apache.geode.distributed.internal.membership.api.CacheOperationMessageMarker;
import org.apache.geode.distributed.internal.membership.api.MemberDisconnectedException;
import org.apache.geode.distributed.internal.membership.api.MemberIdentifier;
import org.apache.geode.distributed.internal.membership.api.MemberShunnedException;
import org.apache.geode.distributed.internal.membership.api.MemberStartupException;
import org.apache.geode.distributed.internal.membership.api.MembershipClosedException;
import org.apache.geode.distributed.internal.membership.api.MembershipConfig;
import org.apache.geode.distributed.internal.membership.api.MembershipConfigurationException;
import org.apache.geode.distributed.internal.membership.api.MembershipStatistics;
import org.apache.geode.distributed.internal.membership.api.Message;
import org.apache.geode.distributed.internal.membership.gms.GMSMemberData;
import org.apache.geode.distributed.internal.membership.gms.GMSMembershipView;
import org.apache.geode.distributed.internal.membership.gms.GMSUtil;
import org.apache.geode.distributed.internal.membership.gms.Services;
import org.apache.geode.distributed.internal.membership.gms.interfaces.HealthMonitor;
import org.apache.geode.distributed.internal.membership.gms.interfaces.Manager;
import org.apache.geode.distributed.internal.membership.gms.interfaces.MessageHandler;
import org.apache.geode.distributed.internal.membership.gms.interfaces.Messenger;
import org.apache.geode.distributed.internal.membership.gms.locator.FindCoordinatorRequest;
import org.apache.geode.distributed.internal.membership.gms.locator.FindCoordinatorResponse;
import org.apache.geode.distributed.internal.membership.gms.messages.JoinRequestMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.JoinResponseMessage;
import org.apache.geode.internal.inet.LocalHostUtil;
import org.apache.geode.internal.serialization.BufferDataOutputStream;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.serialization.StaticSerialization;
import org.apache.geode.internal.serialization.VersionedDataInputStream;
import org.apache.geode.internal.serialization.Versioning;
import org.apache.geode.internal.serialization.VersioningIO;
import org.apache.geode.logging.internal.OSProcess;
import org.apache.geode.util.internal.GeodeGlossary;


/**
 * JGroupsMessenger performs all Membership messaging using a JGroups UDP stack.
 * It uses one of two JGroups configuration files stored in the Geode jar file to
 * configure a simple JGroups stack depending on whether or not multicast is enabled.<br>
 * JGroupsMessenger can send UDP messages using either reliable or non-reliable
 * protocols.<br>
 * Messages must implement the membership Message interface and must be serializable
 * via the serializer/deserializer installed in a Membership.
 */
@SuppressWarnings("StatementWithEmptyBody")
public class JGroupsMessenger<ID extends MemberIdentifier> implements Messenger<ID> {

  private static final Logger logger = Services.getLogger();

  /**
   * The location (in the product) of the non-mcast Jgroups config file.
   */
  private static final String DEFAULT_JGROUPS_TCP_CONFIG =
      "org/apache/geode/distributed/internal/membership/gms/messenger/jgroups-config.xml";

  /**
   * The location (in the product) of the mcast Jgroups config file.
   */
  private static final String JGROUPS_MCAST_CONFIG_FILE_NAME =
      "org/apache/geode/distributed/internal/membership/gms/messenger/jgroups-mcast.xml";

  /** JG magic numbers for types added to the JG ClassConfigurator */
  private static final short JGROUPS_TYPE_JGADDRESS = 2000;
  private static final short JGROUPS_PROTOCOL_TRANSPORT = 1000;

  protected String jgStackConfig;

  JChannel myChannel;
  ID localAddress;
  JGAddress jgAddress;
  private Services<ID> services;

  public JGroupsMessenger() {}

  /** handlers that receive certain classes of messages instead of the Manager */
  private final Map<Class<?>, MessageHandler<?>> handlers = new ConcurrentHashMap<>();

  private volatile GMSMembershipView<ID> view;

  protected final GMSPingPonger pingPonger = new GMSPingPonger();

  protected final AtomicLong pongsReceived = new AtomicLong(0);

  /** tracks multicast messages that have been scheduled for processing */
  protected final Map<ID, MessageTracker> scheduledMcastSeqnos = new HashMap<>();

  protected short nackack2HeaderId;

  /**
   * A set that contains addresses that we have logged JGroups IOExceptions for in the current
   * membership view and possibly initiated suspect processing. This reduces the amount of suspect
   * processing initiated by IOExceptions and the amount of exceptions logged
   */
  private final Set<Address> addressesWithIoExceptionsProcessed =
      Collections.synchronizedSet(new HashSet<>());

  static {
    // register classes that we've added to jgroups that are put on the wire
    // or need a header ID
    ClassConfigurator.add(JGROUPS_TYPE_JGADDRESS, JGAddress.class);
    ClassConfigurator.addProtocol(JGROUPS_PROTOCOL_TRANSPORT, Transport.class);
  }

  private GMSEncrypt<ID> encrypt;

  /**
   * During reconnect a QuorumChecker holds the JGroups channel and responds to Ping
   * and Pong messages but also queues any messages it doesn't recognize. These need
   * to be delivered to handlers after membership services have been rebuilt.
   */
  private Queue<org.jgroups.Message> queuedMessagesFromReconnect;

  /**
   * The JGroupsReceiver is handed messages by the JGroups Channel. It is responsible
   * for deserializating and dispatching those messages to the appropriate handler
   */
  protected JGroupsReceiver jgroupsReceiver;

  public static void setChannelReceiver(JChannel channel, Receiver r) {
    try {
      // Channel.setReceiver() will issue a warning if we try to set a new receiver
      // and the channel already has one. Rather than set the receiver to null &
      // then establish a new one we use reflection to set the channel receiver. See GEODE-7220
      Field receiver = Channel.class.getDeclaredField("receiver");
      receiver.setAccessible(true);
      receiver.set(channel, r);
    } catch (NoSuchFieldException | IllegalAccessException e) {
      throw new IllegalStateException("unable to establish a JGroups receiver", e);
    }
  }

  @Override
  @edu.umd.cs.findbugs.annotations.SuppressWarnings(
      value = "ST_WRITE_TO_STATIC_FROM_INSTANCE_METHOD")
  public void init(Services<ID> s) throws MembershipConfigurationException {
    services = s;

    MembershipConfig config = services.getConfig();

    boolean enableNetworkPartitionDetection = config.isNetworkPartitionDetectionEnabled();
    System.setProperty("jgroups.resolve_dns", String.valueOf(!enableNetworkPartitionDetection));

    InputStream is = null;

    String r;
    if (config.isMulticastEnabled()) {
      r = JGROUPS_MCAST_CONFIG_FILE_NAME;
    } else {
      r = DEFAULT_JGROUPS_TCP_CONFIG;
    }
    ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
    if (contextClassLoader != null) {
      is = contextClassLoader.getResourceAsStream(r);
    }
    if (is == null) {
      is = getClass().getResourceAsStream(r);
    }
    if (is == null) {
      is = ClassLoader.getSystemResourceAsStream(r);
    }
    if (is == null) {
      throw new MembershipConfigurationException(
          String.format("Cannot find %s", r));
    }

    String properties;
    try {
      StringBuilder sb = new StringBuilder(3000);
      BufferedReader br;
      br = new BufferedReader(new InputStreamReader(is, StandardCharsets.US_ASCII));
      String input;
      while ((input = br.readLine()) != null) {
        sb.append(input);
      }
      br.close();
      properties = sb.toString();
    } catch (Exception ex) {
      throw new MembershipConfigurationException(
          "An Exception was thrown while reading JGroups config.",
          ex);
    }

    if (properties.startsWith("<!--")) {
      int commentEnd = properties.indexOf("-->");
      properties = properties.substring(commentEnd + 3);
    }

    if (config.isMulticastEnabled()) {
      properties = replaceStrings(properties, "MCAST_PORT",
          String.valueOf(config.getMcastPort()));
      properties =
          replaceStrings(properties, "MCAST_ADDRESS", config.getMcastAddress());
      properties = replaceStrings(properties, "MCAST_TTL", String.valueOf(config.getMcastTtl()));
      properties = replaceStrings(properties, "MCAST_SEND_BUFFER_SIZE",
          String.valueOf(config.getMcastSendBufferSize()));
      properties = replaceStrings(properties, "MCAST_RECV_BUFFER_SIZE",
          String.valueOf(config.getMcastRecvBufferSize()));
      properties = replaceStrings(properties, "MCAST_RETRANSMIT_INTERVAL", "" + Integer
          .getInteger(GeodeGlossary.GEMFIRE_PREFIX + "mcast-retransmit-interval", 500));
      properties = replaceStrings(properties, "RETRANSMIT_LIMIT",
          String.valueOf(config.getUdpFragmentSize() - 256));
    }

    if (config.isMulticastEnabled() || config.getDisableTcp()
        || (config.getUdpRecvBufferSize() != MembershipConfig.DEFAULT_UDP_RECV_BUFFER_SIZE)) {
      properties =
          replaceStrings(properties, "UDP_RECV_BUFFER_SIZE", "" + config.getUdpRecvBufferSize());
    } else {
      properties = replaceStrings(properties, "UDP_RECV_BUFFER_SIZE",
          "" + MembershipConfig.DEFAULT_UDP_RECV_BUFFER_SIZE_REDUCED);
    }
    properties =
        replaceStrings(properties, "UDP_SEND_BUFFER_SIZE", "" + config.getUdpSendBufferSize());

    String str = config.getBindAddress();
    // JGroups UDP protocol requires a bind address
    if (str == null || str.length() == 0) {
      try {
        str = LocalHostUtil.getLocalHost().getHostAddress();
      } catch (UnknownHostException e) {
        throw new MembershipConfigurationException(e.getMessage(), e);
      }
    }
    properties = replaceStrings(properties, "BIND_ADDR_SETTING", "bind_addr=\"" + str + "\"");

    int port = Integer.getInteger(GeodeGlossary.GEMFIRE_PREFIX + "jg-bind-port", 0);
    if (port != 0) {
      properties = replaceStrings(properties, "MEMBERSHIP_PORT_RANGE_START", "" + port);
      properties = replaceStrings(properties, "MEMBERSHIP_PORT_RANGE", "" + 0);
    } else {
      int[] ports = config.getMembershipPortRange();
      properties = replaceStrings(properties, "MEMBERSHIP_PORT_RANGE_START", "" + ports[0]);
      properties = replaceStrings(properties, "MEMBERSHIP_PORT_RANGE", "" + (ports[1] - ports[0]));
    }

    properties = replaceStrings(properties, "UDP_FRAGMENT_SIZE", "" + config.getUdpFragmentSize());

    properties = replaceStrings(properties, "FC_MAX_CREDITS",
        "" + config.getMcastByteAllowance());
    properties = replaceStrings(properties, "FC_THRESHOLD",
        "" + config.getMcastRechargeThreshold());
    properties = replaceStrings(properties, "FC_MAX_BLOCK",
        "" + config.getMcastRechargeBlockMs());

    jgStackConfig = properties;

    if (!config.getSecurityUDPDHAlgo().isEmpty()) {
      try {
        encrypt = new GMSEncrypt<>(services, config.getSecurityUDPDHAlgo());
        logger.info("Initializing GMSEncrypt ");
      } catch (Exception e) {
        throw new MembershipConfigurationException("problem initializing encryption protocol", e);
      }
    }
  }

  @Override
  @edu.umd.cs.findbugs.annotations.SuppressWarnings(
      value = "ST_WRITE_TO_STATIC_FROM_INSTANCE_METHOD")
  public void start() throws MemberStartupException {
    // create the configuration XML string for JGroups
    String properties = jgStackConfig;

    long start = System.currentTimeMillis();

    // start the jgroups channel and establish the membership ID
    boolean reconnecting = false;
    try {
      Object oldDSMembershipInfo = services.getConfig().getOldMembershipInfo();
      if (oldDSMembershipInfo != null) {
        if (logger.isDebugEnabled()) {
          logger.debug("Reusing JGroups channel from previous system {}", properties);
        }
        MembershipInformationImpl oldInfo = (MembershipInformationImpl) oldDSMembershipInfo;
        myChannel = oldInfo.getChannel();
        queuedMessagesFromReconnect = oldInfo.getQueuedMessages();
        encrypt = oldInfo.getEncrypt();

        // scrub the old channel
        ViewId vid = new ViewId(new JGAddress(), 0);
        List<Address> members = new ArrayList<>();
        members.add(new UUID(0, 0));// TODO open a JGroups JIRA for GEODE-3034
        View jgv = new View(vid, members);
        myChannel.down(new Event(Event.VIEW_CHANGE, jgv));
        // attempt to establish a new UUID in the jgroups channel so the member address will be
        // different
        try {
          Method setAddressMethod = JChannel.class.getDeclaredMethod("setAddress");
          setAddressMethod.setAccessible(true);
          setAddressMethod.invoke(myChannel);
        } catch (SecurityException | NoSuchMethodException e) {
          logger.warn("Unable to establish a new JGroups address.  "
              + "My address will be exactly the same as last time. Exception={}",
              e.getMessage());
        }
        reconnecting = true;
      } else {
        if (logger.isDebugEnabled()) {
          logger.debug("JGroups configuration: {}", properties);
        }

        checkForIPv6();
        InputStream is = new ByteArrayInputStream(properties.getBytes(StandardCharsets.UTF_8));
        myChannel = new JChannel(is);
      }
    } catch (Exception e) {
      throw new MembershipConfigurationException("unable to create jgroups channel", e);
    }

    // give the stats to the jchannel statistics recorder
    StatRecorder<ID> sr =
        (StatRecorder<ID>) myChannel.getProtocolStack().findProtocol(StatRecorder.class);
    if (sr != null) {
      sr.setServices(services);
    }

    Transport<ID> transport = (Transport<ID>) myChannel.getProtocolStack().getTransport();
    transport.setMessenger(this);

    nackack2HeaderId = ClassConfigurator.getProtocolId(NAKACK2.class);

    try {
      jgroupsReceiver = new JGroupsReceiver();
      try {
        setChannelReceiver(myChannel, jgroupsReceiver);
      } catch (IllegalStateException e) {
        throw new MemberStartupException("problem initializing JGroups", e);
      }
      if (!reconnecting) {
        myChannel.connect("AG"); // Apache Geode
      }
    } catch (Exception e) {
      myChannel.close();
      throw new MemberStartupException("unable to create jgroups channel", e);
    }

    establishLocalAddress();

    logger.info("JGroups channel {} (took {}ms)", (reconnecting ? "reinitialized" : "created"),
        System.currentTimeMillis() - start);

  }

  /**
   * JGroups picks an IPv6 address if preferIPv4Stack is false or not set and preferIPv6Addresses is
   * not set or is true. We want it to use an IPv4 address for a dual-IP stack so that both IPv4 and
   * IPv6 messaging work
   */
  private void checkForIPv6() throws Exception {
    boolean preferIpV6Addr = Boolean.getBoolean("java.net.preferIPv6Addresses");
    if (!preferIpV6Addr) {
      if (logger.isDebugEnabled()) {
        logger
            .debug("forcing JGroups to think IPv4 is being used so it will choose an IPv4 address");
      }
      Field m = org.jgroups.util.Util.class.getDeclaredField("ip_stack_type");
      m.setAccessible(true);
      m.set(null, org.jgroups.util.StackType.IPv4);
    }
  }

  @Override
  public void started() throws MemberStartupException {
    if (queuedMessagesFromReconnect != null) {
      logger.info("Delivering {} messages queued by quorum checker",
          queuedMessagesFromReconnect.size());
      for (org.jgroups.Message message : queuedMessagesFromReconnect) {
        jgroupsReceiver.receive(message, true);
      }
      queuedMessagesFromReconnect.clear();
      queuedMessagesFromReconnect = null;
    }
  }

  @Override
  public void stop() {
    if (myChannel != null) {
      if ((services.isShutdownDueToForcedDisconnect() && services.isAutoReconnectEnabled())
          || services.getManager().isReconnectingDS()) {
        // leave the channel open for reconnect attempts
      } else {
        myChannel.close();
      }
    }
  }

  @Override
  public void stopped() {}

  @Override
  public void memberSuspected(ID initiator,
      ID suspect, String reason) {}

  @Override
  public void installView(GMSMembershipView<ID> v) {
    view = v;

    if (jgAddress.getVmViewId() < 0) {
      jgAddress.setVmViewId(localAddress.getVmViewId());
    }
    List<JGAddress> mbrs = v.getMembers().stream().map(JGAddress::new).collect(Collectors.toList());
    ViewId vid = new ViewId(new JGAddress(v.getCoordinator()), v.getViewId());
    View jgv = new View(vid, new ArrayList<>(mbrs));
    if (logger.isTraceEnabled()) {
      logger.trace("installing view into JGroups stack: {}", jgv);
    }
    myChannel.down(new Event(Event.VIEW_CHANGE, jgv));

    addressesWithIoExceptionsProcessed.clear();
    if (encrypt != null) {
      encrypt.installView(v);
    }
    synchronized (scheduledMcastSeqnos) {
      for (ID mbr : v.getCrashedMembers()) {
        scheduledMcastSeqnos.remove(mbr);
      }
      for (ID mbr : v.getShutdownMembers()) {
        scheduledMcastSeqnos.remove(mbr);
      }
    }
  }


  /**
   * If JGroups is unable to send a message it may mean that the network is down. If so we need to
   * initiate suspect processing on the recipient.
   * <p>
   * see Transport._send()
   */
  @SuppressWarnings("UnusedParameters")
  public void handleJGroupsIOException(IOException e, Address dest) {
    if (services.getManager().shutdownInProgress()) { // GEODE-634 - don't log IOExceptions during
                                                      // shutdown
      return;
    }
    GMSMembershipView<ID> v = view;
    JGAddress jgMbr = (JGAddress) dest;
    if (jgMbr != null && v != null) {
      List<ID> members = v.getMembers();
      ID recipient = null;
      for (ID gmsMbr : members) {
        if (jgMbr.getUUIDLsbs() == gmsMbr.getUuidLeastSignificantBits()
            && jgMbr.getUUIDMsbs() == gmsMbr.getUuidMostSignificantBits()
            && jgMbr.getVmViewId() == gmsMbr.getVmViewId()) {
          recipient = gmsMbr;
          break;
        }
      }
      if (recipient != null) {
        if (!addressesWithIoExceptionsProcessed.contains(dest)) {
          logger.warn("Unable to send message to " + recipient, e);
          addressesWithIoExceptionsProcessed.add(dest);
        }
        // If communications aren't working we need to resolve the issue quickly, so here
        // we initiate a final check. Prior to becoming open-source we did a similar check
        // using JGroups VERIFY_SUSPECT
        services.getHealthMonitor().checkIfAvailable(recipient,
            "Unable to send messages to this member via JGroups", true);
      }
    }
  }

  private void establishLocalAddress() throws MemberStartupException {
    UUID logicalAddress = (UUID) myChannel.getAddress();
    logicalAddress = logicalAddress.copy();

    IpAddress ipaddr = (IpAddress) myChannel.down(new Event(Event.GET_PHYSICAL_ADDRESS));

    if (ipaddr != null) {
      jgAddress = new JGAddress(logicalAddress, ipaddr);
    } else {
      UDP udp = (UDP) myChannel.getProtocolStack().getTransport();

      try {
        Method getAddress = UDP.class.getDeclaredMethod("getPhysicalAddress");
        getAddress.setAccessible(true);
        ipaddr = (IpAddress) getAddress.invoke(udp, new Object[0]);
        jgAddress = new JGAddress(logicalAddress, ipaddr);
      } catch (NoSuchMethodException | InvocationTargetException | IllegalAccessException e) {
        throw new MemberStartupException(
            "Unable to configure JGroups channel for membership communications", e);
      }
    }

    // install the address in the JGroups channel protocols
    myChannel.down(new Event(Event.SET_LOCAL_ADDRESS, jgAddress));

    MembershipConfig config = services.getConfig();
    boolean isLocator = (config
        .getVmKind() == MemberIdentifier.LOCATOR_DM_TYPE)
        || !config.getStartLocator().isEmpty();

    // establish the DistributedSystem's address
    String hostname = config.getBindAddress();
    if (hostname == null || hostname.isEmpty()) {
      hostname = jgAddress.getInetAddress().getHostName();
    }
    GMSMemberData gmsMember = new GMSMemberData(jgAddress.getInetAddress(),
        hostname, jgAddress.getPort(),
        OSProcess.getId(), (byte) services.getConfig().getVmKind(),
        -1 /* directport */, -1 /* viewID */, config.getName(),
        GMSUtil.parseGroups(config.getRoles(), config.getGroups()), config.getDurableClientId(),
        config.getDurableClientTimeout(),
        config.isNetworkPartitionDetectionEnabled(), isLocator,
        KnownVersion.getCurrentVersion().ordinal(),
        jgAddress.getUUIDMsbs(), jgAddress.getUUIDLsbs(),
        (byte) (services.getConfig().getMemberWeight() & 0xff), false, null);
    localAddress = services.getMemberFactory().create(gmsMember);
    logger.info("Established local address {}", localAddress);
    services.setLocalAddress(localAddress);
  }

  @Override
  public void beSick() {}

  @Override
  public void playDead() {}

  @Override
  public void beHealthy() {}

  @Override
  public <T extends Message<ID>> void addHandler(Class<T> c, MessageHandler<T> h) {
    handlers.put(c, h);
  }

  @Override
  public boolean testMulticast(long timeout) throws InterruptedException {
    long pongsSnapshot = pongsReceived.longValue();
    JGAddress dest = null;
    try {
      // noinspection ConstantConditions
      pingPonger.sendPingMessage(myChannel, jgAddress, dest);
    } catch (Exception e) {
      logger.warn("unable to send multicast message: {} because of: {}",
          (jgAddress == null ? "multicast recipients" : jgAddress), e.getMessage());
      return false;
    }
    long giveupTime = System.currentTimeMillis() + timeout;
    while (pongsReceived.longValue() == pongsSnapshot && System.currentTimeMillis() < giveupTime) {
      Thread.sleep(100);
    }
    return pongsReceived.longValue() > pongsSnapshot;
  }

  @Override
  public void getMessageState(ID target, Map<String, Long> state,
      boolean includeMulticast) {
    if (includeMulticast) {
      NAKACK2 nakack = (NAKACK2) myChannel.getProtocolStack().findProtocol("NAKACK2");
      if (nakack != null) {
        long seqno = nakack.getCurrentSeqno();
        state.put("JGroups.mcastState", seqno);
      }
    }
  }

  @Override
  public void waitForMessageState(ID sender, Map<String, Long> state)
      throws InterruptedException, TimeoutException {
    Long seqno = state.get("JGroups.mcastState");
    if (seqno == null) {
      return;
    }
    long timeout = services.getConfig().getAckWaitThreshold() * 1000L;
    long startTime = System.currentTimeMillis();
    long warnTime = startTime + timeout;
    long quitTime = warnTime + timeout - 1000L;
    boolean warned = false;

    for (;;) {
      String received = "none";
      long highSeqno = 0;
      synchronized (scheduledMcastSeqnos) {
        MessageTracker tracker = scheduledMcastSeqnos.get(sender);
        if (tracker == null) { // no longer in the membership view
          break;
        }
        highSeqno = tracker.get();
      }

      if (logger.isDebugEnabled()) {
        logger.debug(
            "waiting for multicast messages from {}.  Current seqno={} and expected seqno={}",
            sender, highSeqno, seqno);
      }
      if (highSeqno >= seqno) {
        break;
      }
      long now = System.currentTimeMillis();
      if (!warned && now >= warnTime) {
        warned = true;
        received = String.valueOf(highSeqno);
        logger.warn(
            "{} seconds have elapsed while waiting for multicast messages from {}.  Received {} but expecting at least {}.",
            Long.toString((warnTime - startTime) / 1000L), sender, received, seqno);
      }
      if (now >= quitTime) {
        throw new TimeoutException("Multicast operations from " + sender
            + " did not distribute within " + (now - startTime) + " milliseconds");
      }
      Thread.sleep(50);
    }
  }

  @Override
  public Set<ID> sendUnreliably(Message<ID> msg) {
    return send(msg, false);
  }

  @Override
  public Set<ID> send(Message<ID> msg) {
    return send(msg, true);
  }

  private Set<ID> send(Message<ID> msg, boolean reliably) {

    // perform the same jgroups messaging as in 8.2's GMSMembershipManager.send() method

    // BUT: when marshalling messages we need to include the version of the product and
    // localAddress at the beginning of the message. These should be used in the receiver
    // code to create a versioned input stream, read the sender address, then read the message
    // and set its sender address
    MembershipStatistics theStats = services.getStatistics();
    GMSMembershipView<ID> oldView = view;

    if (!myChannel.isConnected()) {
      logger.info("JGroupsMessenger channel is closed - messaging is not possible");
      throw new MembershipClosedException("Distributed System is shutting down");
    }

    filterOutgoingMessage(msg);

    List<ID> destinations = msg.getRecipients();
    boolean allDestinations = msg.forAll();

    boolean useMcast = false;
    if (services.getConfig().isMulticastEnabled()) {
      if (msg.getMulticast() || allDestinations) {
        useMcast = services.getManager().isMulticastAllowed();
      }
    }

    if (logger.isDebugEnabled() && reliably) {
      String recips = useMcast ? "multicast" : destinations.toString();
      if (logger.isDebugEnabled()) {
        logger.debug("sending via JGroups: [{}] recipients: {}", msg, recips);
      }
    }

    JGAddress local = jgAddress;

    Set<ID> failedRecipients = new HashSet<>();
    if (useMcast) {
      long startSer = theStats.startMsgSerialization();
      org.jgroups.Message jmsg;
      try {
        jmsg =
            createJGMessage(msg, local, null, KnownVersion.getCurrentVersion().ordinal());
      } catch (IOException e) {
        return new HashSet<>(msg.getRecipients());
      } finally {
        theStats.endMsgSerialization(startSer);
      }

      Exception problem;
      try {
        jmsg.setTransientFlag(TransientFlag.DONT_LOOPBACK);
        if (!reliably) {
          jmsg.setFlag(org.jgroups.Message.Flag.NO_RELIABILITY);
        }
        theStats.incSentBytes(jmsg.getLength());
        if (logger.isTraceEnabled()) {
          logger.trace("Sending JGroups message: {}", jmsg);
        }
        myChannel.send(jmsg);
      } catch (Exception e) {
        if (logger.isDebugEnabled()) {
          logger.debug("caught unexpected exception", e);
        }
        Throwable cause = e.getCause();
        if (cause instanceof MemberDisconnectedException) {
          problem = (Exception) cause;
        } else {
          problem = e;
        }
        if (services.getShutdownCause() != null) {
          Throwable shutdownCause = services.getShutdownCause();
          // If ForcedDisconnectException occurred then report it as actual
          // problem.
          if (shutdownCause instanceof MemberDisconnectedException) {
            problem = (Exception) shutdownCause;
          } else {
            Throwable ne = problem;
            while (ne.getCause() != null) {
              ne = ne.getCause();
            }
            ne.initCause(services.getShutdownCause());
          }
        }
        final String channelClosed =
            "Channel closed";
        throw new MembershipClosedException(channelClosed, problem);
      }
    } // useMcast
    else { // ! useMcast
      int len = destinations.size();
      List<ID> calculatedMembers; // explicit list of members
      int calculatedLen; // == calculatedMembers.len
      if (len == 1 && destinations.get(0) == ALL_RECIPIENTS) { // send to all
        // Grab a copy of the current membership
        GMSMembershipView<ID> v = services.getJoinLeave().getView();

        // Construct the list
        calculatedLen = v.size();
        calculatedMembers = new LinkedList<>();
        for (int i = 0; i < calculatedLen; i++) {
          ID m = v.get(i);
          calculatedMembers.add(m);
        }
      } // send to all
      else { // send to explicit list
        calculatedLen = len;
        calculatedMembers = new LinkedList<>();
        for (int i = 0; i < calculatedLen; i++) {
          calculatedMembers.add(destinations.get(i));
        }
      } // send to explicit list
      Int2ObjectOpenHashMap<org.jgroups.Message> messages = new Int2ObjectOpenHashMap<>();
      long startSer = theStats.startMsgSerialization();
      boolean firstMessage = true;
      for (ID mbr : calculatedMembers) {
        short version = mbr.getVersionOrdinal();
        if (!messages.containsKey(version)) {
          org.jgroups.Message jmsg;
          try {
            jmsg = createJGMessage(msg, local, mbr, version);
            messages.put(version, jmsg);
          } catch (IOException e) {
            failedRecipients.add(mbr);
            continue;
          }
          if (firstMessage) {
            theStats.incSentBytes(jmsg.getLength());
            firstMessage = false;
          }
        }
      }
      theStats.endMsgSerialization(startSer);
      Collections.shuffle(calculatedMembers);
      int i = 0;
      for (ID mbr : calculatedMembers) {
        JGAddress to = new JGAddress(mbr);
        short version = mbr.getVersionOrdinal();
        org.jgroups.Message jmsg = messages.get(version);
        if (jmsg == null) {
          continue; // failed for all recipients
        }
        Exception problem = null;
        try {
          org.jgroups.Message tmp = (i < (calculatedLen - 1)) ? jmsg.copy(true) : jmsg;
