<commit_before>/*
 * Cacheonix systems licenses this file to You under the LGPL 2.1
 * (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.cacheonix.org/products/cacheonix/license-lgpl-2.1.htm
 *
 * 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.cacheonix.impl.net.cluster;

import java.util.Collections;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.Executor;

import org.cacheonix.cluster.ClusterEventSubscriber;
import org.cacheonix.cluster.ClusterState;
import org.cacheonix.impl.cluster.ClusterStateChangedEventImpl;
import org.cacheonix.impl.net.ClusterNodeAddress;
import org.cacheonix.impl.net.processor.Frame;
import org.cacheonix.impl.net.processor.InvalidMessageException;
import org.cacheonix.impl.net.processor.ReceiverAddress;
import org.cacheonix.impl.net.processor.Request;
import org.cacheonix.impl.net.processor.Response;
import org.cacheonix.impl.net.processor.UUID;
import org.cacheonix.impl.util.Assert;
import org.cacheonix.impl.util.CollectionUtils;
import org.cacheonix.impl.util.logging.Logger;

import static org.cacheonix.impl.cluster.ClusterEventUtil.convertStateMachineToUserClusterState;
import static org.cacheonix.impl.net.cluster.ClusterProcessorState.STATE_RECOVERY;
import static org.cacheonix.impl.net.processor.Response.RESULT_SUCCESS;

/**
 * Marker request is a marker that sends
 */
public abstract class MarkerRequest extends ClusterRequest {

   /**
    * Logger.
    *
    * @noinspection UNUSED_SYMBOL, UnusedDeclaration
    */
   private static final Logger LOG = Logger.getLogger(MarkerRequest.class); // NOPMD


   protected MarkerRequest(final int wireableType) {

      super(wireableType);
   }


   /**
    * {@inheritDoc}
    * <p/>
    * This implementation enhances the default behaviour by incrementing number of markers received by a cluster node.
    */
   public void execute() throws InterruptedException {

      //
      final ClusterProcessor processor = getClusterProcessor();

      // Increment marker counter

      processor.getProcessorState().incrementMarkerCounter();

      // Stop timeout
      processor.cancelMarkerTimeout();

      try {

         super.execute();
      } finally {

         processor.cancelMarkerTimeout();
      }
   }


   public void validate() throws InvalidMessageException {

      super.validate();

      if (!isSenderSet()) {
         throw new InvalidMessageException("Marker does not have a receiver");
      }

      if (!isReceiverSet()) {
         throw new InvalidMessageException("Marker does not have a receiver");
      }
   }


   /**
    * Sends a left mcast announcement to self to support ordering of configuration changes with messages (a total
    * order).
    *
    * @param reservedLeftSeqNum a seqNum supplied by the leaving node.
    * @param left               the address of the leaving node.
    */
   protected final void sendLeftToSelf(final long reservedLeftSeqNum, final ClusterNodeAddress left) {

      //noinspection ControlFlowStatementWithoutBraces
      if (LOG.isDebugEnabled())
         LOG.debug("Sending left to self, reservedLeftSeqNum: " + reservedLeftSeqNum + ", left: " + left); // NOPMD

      final ClusterProcessor processor = getClusterProcessor();

      // Create message
      final ClusterNodeLeftAnnouncement nodeLeftMessage = new ClusterNodeLeftAnnouncement();

      nodeLeftMessage.setClusterUUID(processor.getProcessorState().getClusterView().getClusterUUID());
      nodeLeftMessage.setSender(left);
      nodeLeftMessage.setLeave(left);

      // Convert to a frame
      final PayloadPartitioner payloadPartitioner = new PayloadPartitioner();
      final List<Frame> frames = payloadPartitioner.partition(nodeLeftMessage);
      Assert.assertTrue(frames.size() == 1, "ClusterNodeLeftAnnouncement should be partitioned to a single frame");

      // Set reserved sequence number
      final Frame nodeLeftFrame = frames.get(0);
      nodeLeftFrame.setSequenceNumber(reservedLeftSeqNum);

      nodeLeftFrame.setClusterUUID(processor.getProcessorState().getClusterView().getClusterUUID());

      // Send to ourselves

      processor.getProcessorState().getReceivedList().add(nodeLeftFrame);
   }


   /**
    * Sends a join mcast announcement to self to support ordering of configuration changes with messages (a total
    * order).
    *
    * @param reservedJoinSeqNum a seqNum supplied by the join coordinator (AKA predecessor).
    * @param joined             the address of the joined node.
    */<commit_msg>Fixed not setting a timestamp on the announcement<commit_after>/*
 * Cacheonix systems licenses this file to You under the LGPL 2.1
 * (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.cacheonix.org/products/cacheonix/license-lgpl-2.1.htm
 *
 * 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.cacheonix.impl.net.cluster;

import java.util.Collections;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.Executor;

import org.cacheonix.cluster.ClusterEventSubscriber;
import org.cacheonix.cluster.ClusterState;
import org.cacheonix.impl.cluster.ClusterStateChangedEventImpl;
import org.cacheonix.impl.net.ClusterNodeAddress;
import org.cacheonix.impl.net.processor.Frame;
import org.cacheonix.impl.net.processor.InvalidMessageException;
import org.cacheonix.impl.net.processor.ReceiverAddress;
import org.cacheonix.impl.net.processor.Request;
import org.cacheonix.impl.net.processor.Response;
import org.cacheonix.impl.net.processor.UUID;
import org.cacheonix.impl.util.Assert;
import org.cacheonix.impl.util.CollectionUtils;
import org.cacheonix.impl.util.logging.Logger;

import static org.cacheonix.impl.cluster.ClusterEventUtil.convertStateMachineToUserClusterState;
import static org.cacheonix.impl.net.cluster.ClusterProcessorState.STATE_RECOVERY;
import static org.cacheonix.impl.net.processor.Response.RESULT_SUCCESS;

/**
 * Marker request is a marker that sends
 */
public abstract class MarkerRequest extends ClusterRequest {

   /**
    * Logger.
    *
    * @noinspection UNUSED_SYMBOL, UnusedDeclaration
    */
   private static final Logger LOG = Logger.getLogger(MarkerRequest.class); // NOPMD


   protected MarkerRequest(final int wireableType) {

      super(wireableType);
   }


   /**
    * {@inheritDoc}
    * <p/>
    * This implementation enhances the default behaviour by incrementing number of markers received by a cluster node.
    */
   public void execute() throws InterruptedException {

      //
      final ClusterProcessor processor = getClusterProcessor();

      // Increment marker counter

      processor.getProcessorState().incrementMarkerCounter();

      // Stop timeout
      processor.cancelMarkerTimeout();

      try {

         super.execute();
      } finally {

         processor.cancelMarkerTimeout();
      }
   }


   public void validate() throws InvalidMessageException {

      super.validate();

      if (!isSenderSet()) {
         throw new InvalidMessageException("Marker does not have a receiver");
      }

      if (!isReceiverSet()) {
         throw new InvalidMessageException("Marker does not have a receiver");
      }
   }


   /**
    * Sends a left mcast announcement to self to support ordering of configuration changes with messages (a total
    * order).
    *
    * @param reservedLeftSeqNum a seqNum supplied by the leaving node.
    * @param left               the address of the leaving node.
    */
   protected final void sendLeftToSelf(final long reservedLeftSeqNum, final ClusterNodeAddress left) {

      //noinspection ControlFlowStatementWithoutBraces
      if (LOG.isDebugEnabled())
         LOG.debug("Sending left to self, reservedLeftSeqNum: " + reservedLeftSeqNum + ", left: " + left); // NOPMD

      final ClusterProcessor processor = getClusterProcessor();

      // Create message
      final ClusterNodeLeftAnnouncement nodeLeftMessage = new ClusterNodeLeftAnnouncement();
      nodeLeftMessage.setClusterUUID(processor.getProcessorState().getClusterView().getClusterUUID());
      nodeLeftMessage.setTimestamp(processor.getClock().currentTime());
      nodeLeftMessage.setSender(left);
      nodeLeftMessage.setLeave(left);

      // Convert to a frame
      final PayloadPartitioner payloadPartitioner = new PayloadPartitioner();
      final List<Frame> frames = payloadPartitioner.partition(nodeLeftMessage);
      Assert.assertTrue(frames.size() == 1, "ClusterNodeLeftAnnouncement should be partitioned to a single frame");

      // Set reserved sequence number
      final Frame nodeLeftFrame = frames.get(0);
      nodeLeftFrame.setSequenceNumber(reservedLeftSeqNum);

      nodeLeftFrame.setClusterUUID(processor.getProcessorState().getClusterView().getClusterUUID());

      // Send to ourselves

      processor.getProcessorState().getReceivedList().add(nodeLeftFrame);
   }


   /**
    * Sends a join mcast announcement to self to support ordering of configuration changes with messages (a total
    * order).
    *
    * @param reservedJoinSeqNum a seqNum supplied by the join coordinator (AKA predecessor).
    * @param joined             the address of the joined node.
    */
{'fraction_non_alphanumeric': 0.08725579512096367, 'fraction_numerical': 0.0012146978439113271, 'mean_word_length': 3.6319737458977963, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3103803', 'n_tokens_mistral': 2624, 'n_tokens_neox': 2544, 'n_words': 740}