public void distribute(EnumListenerEvent operation, EntryEventImpl event,
      List<Integer> allRemoteDSIds, boolean isLastEventInTransaction) {

    final boolean isDebugEnabled = logger.isDebugEnabled();
    boolean wasInterrupted = false;

    // released by this method or transfers ownership to TmpQueueEvent
    @Released
    EntryEventImpl clonedEvent = new EntryEventImpl(event, false);
    boolean freeClonedEvent = true;
    try {

      final GatewaySenderStats stats = getStatistics();
      stats.incEventsReceived();

      if (!checkForDistribution(event, stats)) {
        stats.incEventsNotQueued();
        return;
      }

      // this filter is defined by Asif which exist in old wan too. new wan has
      // other GatewaEventFilter. Do we need to get rid of this filter. Cheetah is
      // not considering this filter
      if (!filter.enqueueEvent(event)) {
        stats.incEventsFiltered();
        return;
      }

      // start to distribute
      setModifiedEventId(clonedEvent);
      Object callbackArg = clonedEvent.getRawCallbackArgument();

      if (isDebugEnabled) {
        // We can't deserialize here for logging purposes so don't
        // call getNewValue.
        // event.getNewValue(); // to deserialize the value if necessary
        logger.debug("{} : About to notify {} to perform operation {} for {} callback arg {}",
            isPrimary(), getId(), operation, clonedEvent, callbackArg);
      }

      if (callbackArg instanceof GatewaySenderEventCallbackArgument) {
        GatewaySenderEventCallbackArgument seca = (GatewaySenderEventCallbackArgument) callbackArg;
        if (isDebugEnabled) {
          logger.debug(
              "{}: Event originated in {}. My DS id is {}. The remote DS id is {}. The recipients are: {}",
              this, seca.getOriginatingDSId(), getMyDSId(), getRemoteDSId(),
              seca.getRecipientDSIds());
        }
        if (seca.getOriginatingDSId() == DEFAULT_DISTRIBUTED_SYSTEM_ID) {
          if (isDebugEnabled) {
            logger.debug(
                "{}: Event originated in {}. My DS id is {}. The remote DS id is {}. The recipients are: {}",
                this, seca.getOriginatingDSId(), getMyDSId(), getRemoteDSId(),
                seca.getRecipientDSIds());
          }

          seca.setOriginatingDSId(getMyDSId());
          seca.initializeReceipientDSIds(allRemoteDSIds);

        } else {
          // if the dispatcher is GatewaySenderEventCallbackDispatcher (which is the case of WBCL),
          // skip the below check of remoteDSId.
          // Fix for #46517
          AbstractGatewaySenderEventProcessor ep = getEventProcessor();
          // if manual-start is true, ep is null
          if (ep == null || !(ep.getDispatcher() instanceof GatewaySenderEventCallbackDispatcher)) {
            if (seca.getOriginatingDSId() == getRemoteDSId()) {
              if (isDebugEnabled) {
                logger.debug(
                    "{}: Event originated in {}. My DS id is {}. It is being dropped as remote is originator.",
                    this, seca.getOriginatingDSId(), getMyDSId());
              }
              return;
            } else if (seca.getRecipientDSIds().contains(getRemoteDSId())) {
              if (isDebugEnabled) {
                logger.debug(
                    "{}: Event originated in {}. My DS id is {}. The remote DS id is {}.. It is being dropped as remote ds is already a recipient. Recipients are: {}",
                    this, seca.getOriginatingDSId(), getMyDSId(), getRemoteDSId(),
                    seca.getRecipientDSIds());
              }
              return;
            }
          }
          seca.getRecipientDSIds().addAll(allRemoteDSIds);
        }
      } else {
        GatewaySenderEventCallbackArgument geCallbackArg =
            new GatewaySenderEventCallbackArgument(callbackArg, getMyDSId(), allRemoteDSIds);
        clonedEvent.setCallbackArgument(geCallbackArg);
      }

      // If this gateway is not running, return
      if (!getIsRunningAndDropEventIfNotRunning(event, isDebugEnabled, clonedEvent)) {
        return;
      }

      if (AbstractGatewaySender.doSleepForTestingInDistribute.get()) {
        try {
          Thread.sleep(5);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
      if (!getLifeCycleLock().readLock().tryLock()) {
        synchronized (queuedEventsSync) {
          if (!enqueuedAllTempQueueEvents) {
            if (!getLifeCycleLock().readLock().tryLock()) {
              Object substituteValue = getSubstituteValue(clonedEvent, operation);
              tmpQueuedEvents.add(new TmpQueueEvent(operation, clonedEvent, substituteValue));
              freeClonedEvent = false;
              stats.incTempQueueSize();
              if (isDebugEnabled) {
                logger.debug("Event : {} is added to TempQueue", clonedEvent);
              }
              return;
            }
          }
        }
        if (enqueuedAllTempQueueEvents) {
          while (true) {
            try {
              while (!getLifeCycleLock().readLock().tryLock(10, TimeUnit.MILLISECONDS)) {
                if (!getIsRunningAndDropEventIfNotRunning(event, isDebugEnabled, clonedEvent)) {
                  return;
                }
              }
              break;
            } catch (InterruptedException e) {
              wasInterrupted = true;
            }
          }
        }
      }
      try {
        // If this gateway is not running, return
        // The sender may have stopped, after we have checked the status in the beginning.
        if (!getIsRunningAndDropEventIfNotRunning(event, isDebugEnabled, clonedEvent)) {
          return;
        }

        try {
          AbstractGatewaySenderEventProcessor ev = eventProcessor;
          if (ev == null) {
            getStopper().checkCancelInProgress(null);
            getCache().getDistributedSystem().getCancelCriterion().checkCancelInProgress(null);
            // event processor will be null if there was an authorization
            // problem
            // connecting to the other site (bug #40681)
            if (ev == null) {
              throw new GatewayCancelledException("Event processor thread is gone");
            }
          }

          // Get substitution value to enqueue if necessary
          Object substituteValue = getSubstituteValue(clonedEvent, operation);

          ev.enqueueEvent(operation, clonedEvent, substituteValue, isLastEventInTransaction);
        } catch (CancelException e) {
          logger.debug("caught cancel exception", e);
          throw e;
        } catch (RegionDestroyedException e) {
          logger.warn(String.format(
              "%s: An Exception occurred while queueing %s to perform operation %s for %s",
              this, getId(), operation, clonedEvent),
              e);
        } catch (Exception e) {
          logger.fatal(String.format(
              "%s: An Exception occurred while queueing %s to perform operation %s for %s",
              this, getId(), operation, clonedEvent),
              e);
        }
      } finally {
        getLifeCycleLock().readLock().unlock();
      }
    } finally {
      if (freeClonedEvent) {
        clonedEvent.release(); // fix for bug 48035
      }
      if (wasInterrupted) {
        Thread.currentThread().interrupt();
      }
    }
  }
Check if this event can be distributed by senders.

@return boolean True if the event is allowed.
geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java