@Override
  public Set<InternalDistributedMember> directChannelSend(
      List<InternalDistributedMember> destinations,
      DistributionMessage content)
      throws NotSerializableException {
    MembershipStatistics theStats = clusterDistributionManager.getStats();
    boolean allDestinations;
    InternalDistributedMember[] keys;
    if (content.forAll()) {
      allDestinations = true;
      keys = membership.getAllMembers(EMPTY_MEMBER_ARRAY);
    } else {
      allDestinations = false;
      keys = destinations.toArray(EMPTY_MEMBER_ARRAY);
    }

    int sentBytes;
    try {
      sentBytes =
          directChannel.send(membership, keys, content, ackWaitThreshold,
              ackSevereAlertThreshold);

      if (theStats != null) {
        theStats.incSentBytes(sentBytes);
      }

      if (sentBytes == 0) {
        checkCancelled();
      }
    } catch (MembershipClosedException e) {
      checkCancelled();
      throw new DistributedSystemDisconnectedException(e.getMessage(), e.getCause());
    } catch (DistributedSystemDisconnectedException ex) {
      checkCancelled();
      throw ex;
    } catch (ConnectExceptions ex) {
      // Check if the connect exception is due to system shutting down.
      if (membership.shutdownInProgress()) {
        checkCancelled();
        throw new DistributedSystemDisconnectedException();
      }

      if (allDestinations) {
        return null;
      }

      // We need to return this list of failures
      List<InternalDistributedMember> members = ex.getMembers();

      // SANITY CHECK: If we fail to send a message to an existing member
      // of the view, we have a serious error (bug36202).


      // Iterate through members and causes in tandem :-(
      Iterator<InternalDistributedMember> it_mem = members.iterator();
      Iterator<Throwable> it_causes = ex.getCauses().iterator();
      while (it_mem.hasNext()) {
        InternalDistributedMember member = it_mem.next();
        Throwable th = it_causes.next();

        if (!membership.hasMember(member) || (th instanceof ShunnedMemberException)) {
          if (logger.isDebugEnabled()) {
            logger
                .debug(String.format("Failed to send message <%s> to member <%s> no longer in view",
                    content.getShortClassName(), member), th);
          }
          continue;
        }
        logger
            .fatal(String.format("Failed to send message <%s> to member <%s> view, %s",
                // TODO - This used to be services.getJoinLeave().getView(), which is a different
                // view object. Is it ok to log membershipManager.getView here?
                content, member, membership.getView()),
                th);
      }
      return new HashSet<>(members);
    } // catch ConnectionExceptions
    catch (ToDataException | CancelException e) {
      checkCancelled();
      throw e;
    } catch (NotSerializableException | RuntimeException | Error e) {
      if (logger.isDebugEnabled()) {
        logger.debug("Membership: directChannelSend caught exception: {}",
            e.getMessage(), e);
      }
      checkCancelled();
      throw e;
    }
    return null;
  }
Perform the grossness associated with sending a message over a DirectChannel

@param destinations the list of destinations
@param content the message
@return all recipients who did not receive the message (null if all received it)
@throws NotSerializableException if the message is not serializable
geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionImpl.java