content: /*
   * 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.cassandra.net;
  
  import java.io.IOException;
  import java.nio.ByteBuffer;
  import java.util.Collections;
  import java.util.EnumMap;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.function.Supplier;
  import javax.annotation.Nullable;
  
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.primitives.Ints;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.cassandra.config.DatabaseDescriptor;
  import org.apache.cassandra.exceptions.RequestFailure;
  import org.apache.cassandra.exceptions.RequestFailureReason;
  import org.apache.cassandra.io.IVersionedAsymmetricSerializer;
  import org.apache.cassandra.io.IVersionedSerializer;
  import org.apache.cassandra.io.util.DataInputBuffer;
  import org.apache.cassandra.io.util.DataInputPlus;
  import org.apache.cassandra.io.util.DataOutputPlus;
  import org.apache.cassandra.locator.InetAddressAndPort;
  import org.apache.cassandra.tcm.ClusterMetadata;
  import org.apache.cassandra.tcm.Epoch;
  import org.apache.cassandra.tracing.Tracing;
  import org.apache.cassandra.tracing.Tracing.TraceType;
  import org.apache.cassandra.transport.Dispatcher;
  import org.apache.cassandra.utils.MonotonicClockTranslation;
  import org.apache.cassandra.utils.NoSpamLogger;
  import org.apache.cassandra.utils.TimeUUID;
  
  import static java.util.concurrent.TimeUnit.MINUTES;
  import static java.util.concurrent.TimeUnit.NANOSECONDS;
  import static org.apache.cassandra.db.TypeSizes.sizeof;
  import static org.apache.cassandra.db.TypeSizes.sizeofUnsignedVInt;
  import static org.apache.cassandra.net.MessagingService.VERSION_40;
  import static org.apache.cassandra.net.MessagingService.VERSION_50;
  import static org.apache.cassandra.net.MessagingService.VERSION_51;
  import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
  import static org.apache.cassandra.utils.MonotonicClock.Global.approxTime;
  import static org.apache.cassandra.utils.vint.VIntCoding.computeUnsignedVIntSize;
  import static org.apache.cassandra.utils.vint.VIntCoding.getUnsignedVInt;
  import static org.apache.cassandra.utils.vint.VIntCoding.getUnsignedVInt32;
  import static org.apache.cassandra.utils.vint.VIntCoding.skipUnsignedVInt;
  
  /**
   * Immutable main unit of internode communication - what used to be {@code MessageIn} and {@code MessageOut} fused
   * in one class.
   *
   * @param <T> The type of the message payload.
   */
  public class Message<T> implements ResponseContext
  {
      private static final Logger logger = LoggerFactory.getLogger(Message.class);
      private static final NoSpamLogger noSpam1m = NoSpamLogger.getLogger(logger, 1, TimeUnit.MINUTES);
  
      private static final Supplier<Epoch> epochSupplier = () -> ClusterMetadata.current().epoch;
  
      public final Header header;
      public final T payload;
  
      private final IVersionedAsymmetricSerializer<T, ?> payloadSerializer;
  
      Message(Header header, T payload)
      {
          this.header = header;
          this.payload = payload;
          this.payloadSerializer = verb().serializer();
      }
  
      /** Whether the message has crossed the node boundary, that is whether it originated from another node. */
      public boolean isCrossNode()
      {
          return !from().equals(getBroadcastAddressAndPort());
      }
  
      /** Sender of the message. */
      @Override
      public InetAddressAndPort from()
      {
          return header.from;
      }
  
      /**
       * id of the request/message. In 4.0+ can be shared between multiple messages of the same logical request,
       * whilst in versions above a new id would be allocated for each message sent.
       */
      @Override
      public long id()
      {
          return header.id;
      }
  
      public Epoch epoch()
      {
          return header.epoch;
      }
  
      @Override
      public Verb verb()
      {
          return header.verb;
      }
  
      public boolean isFailureResponse()
      {
          return verb() == Verb.FAILURE_RSP;
      }
  
      /**
       * Creation time of the message. If cross-node timeouts are enabled ({@link DatabaseDescriptor#hasCrossNodeTimeout()},
       * {@code deserialize()} will use the marshalled value, otherwise will use current time on the deserializing machine.
       */
      public long createdAtNanos()
      {
          return header.createdAtNanos;
      }
  
      @Override
      public long expiresAtNanos()
      {
          return header.expiresAtNanos;
      }
  
      /** For how long the message has lived. */
      public long elapsedSinceCreated(TimeUnit units)
      {
          return units.convert(approxTime.now() - createdAtNanos(), NANOSECONDS);
      }
  
      public long creationTimeMillis()
      {
          return approxTime.translate().toMillisSinceEpoch(createdAtNanos());
      }
  
      /** Whether a failure response should be returned upon failure */
      boolean callBackOnFailure()
      {
          return header.callBackOnFailure();
      }
  
      public boolean trackWarnings()
      {
          return header.trackWarnings();
      }
  
      /** See CASSANDRA-14145 */
      public boolean trackRepairedData()
      {
          return header.trackRepairedData();
      }
  
      /** Used for cross-DC write optimisation - pick one node in the DC and have it relay the write to its local peers */
      @Nullable
      public ForwardingInfo forwardTo()
      {
          return header.forwardTo();
      }
  
      /** The originator of the request - used when forwarding and will differ from {@link #from()} */
      @Nullable
      public InetAddressAndPort respondTo()
      {
          return header.respondTo();
      }
  
      @Nullable
      public TimeUUID traceSession()
      {
          return header.traceSession();
      }
  
      @Nullable
      public TraceType traceType()
      {
          return header.traceType();
      }
  
      /*
       * request/response convenience
       */
  
      /**
       * Make a request {@link Message} with supplied verb and payload. Will fill in remaining fields
       * automatically.
       *
       * If you know that you will need to set some params or flags - prefer using variants of {@code out()}
       * that allow providing them at point of message constructions, rather than allocating new messages
       * with those added flags and params. See {@code outWithFlag()}, {@code outWithFlags()}, and {@code outWithParam()}
       * family.
       */
      public static <T> Message<T> out(Verb verb, T payload)
      {
          assert !verb.isResponse() : verb;
  
          return outWithParam(nextId(), verb, payload, null, null);
      }
  
      public static <T> Message<T> synthetic(InetAddressAndPort from, Verb verb, T payload)
      {
          return new Message<>(new Header(-1, epochSupplier.get(), verb, from, -1, -1, 0, NO_PARAMS), payload);
      }
  
      public static <T> Message<T> out(Verb verb, T payload, long expiresAtNanos)
      {
          return outWithParam(nextId(), verb, expiresAtNanos, payload, 0, null, null);
      }
  
      public static <T> Message<T> out(Verb verb, T payload, boolean isUrgent)
      {
          assert !verb.isResponse();
          if (isUrgent)
              return outWithFlag(verb, payload,  MessageFlag.URGENT);
          else
              return out(verb, payload);
      }
  
      public static <T> Message<T> outWithFlag(Verb verb, T payload, MessageFlag flag)
      {
          assert !verb.isResponse();
          return outWithParam(nextId(), verb, 0, payload, flag.addTo(0), null, null);
      }
  
      public static <T> Message<T> outWithFlags(Verb verb, T payload, MessageFlag flag1, MessageFlag flag2)
      {
          assert !verb.isResponse();
          return outWithParam(nextId(), verb, 0, payload, flag2.addTo(flag1.addTo(0)), null, null);
      }
  
      public static <T> Message<T> outWithFlags(Verb verb, T payload, Dispatcher.RequestTime requestTime, List<MessageFlag> flags)
      {
          assert !verb.isResponse();
          int encodedFlags = 0;
          for (MessageFlag flag : flags)
              encodedFlags = flag.addTo(encodedFlags);
  
          return new Message<T>(new Header(nextId(),
                                           epochSupplier.get(),
                                           verb,
                                           getBroadcastAddressAndPort(),
                                           requestTime.startedAtNanos(),
                                           requestTime.computeDeadline(verb.expiresAfterNanos()),
                                           encodedFlags,
                                           buildParams(null, null)),
                                payload);
      }
  
      @VisibleForTesting
      static <T> Message<T> outWithParam(long id, Verb verb, T payload, ParamType paramType, Object paramValue)
      {
          return outWithParam(id, verb, 0, payload, paramType, paramValue);
      }
  
      private static <T> Message<T> outWithParam(long id, Verb verb, long expiresAtNanos, T payload, ParamType paramType, Object paramValue)
      {
          return outWithParam(id, verb, expiresAtNanos, payload, 0, paramType, paramValue);
      }
  
      private static <T> Message<T> outWithParam(long id, Verb verb, long expiresAtNanos, T payload, int flags, ParamType paramType, Object paramValue)
      {
          return withParam(getBroadcastAddressAndPort(), id, verb, expiresAtNanos, payload, flags, paramType, paramValue);
      }
  
      private static <T> Message<T> withParam(InetAddressAndPort from, long id, Verb verb, long expiresAtNanos, T payload, int flags, ParamType paramType, Object paramValue)
      {
          if (payload == null)
              throw new IllegalArgumentException();
  
          long createdAtNanos = approxTime.now();
          if (expiresAtNanos == 0)
              expiresAtNanos = verb.expiresAtNanos(createdAtNanos);
  
          return new Message<>(new Header(id, epochSupplier.get(), verb, from, createdAtNanos, expiresAtNanos, flags, buildParams(paramType, paramValue)), payload);
      }
  
      public static <T> Message<T> internalResponse(Verb verb, T payload)
      {
          assert verb.isResponse();
          return outWithParam(0, verb, payload, null, null);
      }
  
      /**
       * Used by the {@code MultiRangeReadCommand} to split multi-range responses from a replica
       * into single-range responses.
       */
      @VisibleForTesting
      public static <T> Message<T> remoteResponse(InetAddressAndPort from, Verb verb, T payload)
      {
          assert verb.isResponse();
          long createdAtNanos = approxTime.now();
          long expiresAtNanos = verb.expiresAtNanos(createdAtNanos);
          return new Message<>(new Header(0, epochSupplier.get(), verb, from, createdAtNanos, expiresAtNanos, 0, NO_PARAMS), payload);
      }
  
      /**
       * A way to generate messages originating from arbitrary node. Should ONLY be used for testing purposes, as meddling with an
       * originator can be dangerous.
       */
      @VisibleForTesting
      public static <T> Message<T> remoteResponseForTests(long id, InetAddressAndPort from, Verb verb, T payload)
      {
          assert verb.isResponse();
          long createdAtNanos = approxTime.now();
          long expiresAtNanos = verb.expiresAtNanos(createdAtNanos);
          return new Message<>(new Header(id, epochSupplier.get(), verb, from, createdAtNanos, expiresAtNanos, 0, NO_PARAMS), payload);
      }
  
      @VisibleForTesting
      public static <T> Message<T> forgeIdentityForTests(Message<T> msg, InetAddressAndPort addr)
      {
          return new Message<>(new Header(msg.header.id,
                                          msg.header.epoch,
                                          msg.header.verb,
                                          addr,
                                          msg.header.createdAtNanos,
                                          msg.header.expiresAtNanos,
                                          msg.header.flags,
                                          msg.header.params),
                               msg.payload);
      }
  
      /** Builds a response Message with provided payload, and all the right fields inferred from request Message */
      public <T> Message<T> responseWith(T payload)
      {
          Message<T> msg = responseWith(payload, this);
          if (header.hasFlag(MessageFlag.URGENT))
              msg = msg.withFlag(MessageFlag.URGENT);
          return msg;
      }
  
      public static <T> Message<T> responseWith(T payload, ResponseContext respondTo)
      {
          return outWithParam(respondTo.id(), respondTo.verb().responseVerb, respondTo.expiresAtNanos(), payload, null, null);
      }
  
      /** Builds a response Message with no payload, and all the right fields inferred from request Message */
      public Message<NoPayload> emptyResponse()
      {
          return responseWith(NoPayload.noPayload);
      }
  
      /** Builds a failure response Message with an explicit reason, and fields inferred from request Message */
      public Message<RequestFailure> failureResponse(RequestFailureReason reason)
      {
          return failureResponse(reason, null);
      }
  
      public Message<RequestFailure> failureResponse(RequestFailureReason reason, @Nullable Throwable failure)
      {
          return failureResponse(reason, failure, this);
      }
  
      public static Message<RequestFailure> failureResponse(RequestFailureReason reason, @Nullable Throwable failure, ResponseContext respondTo)
      {
          return failureResponse(respondTo.id(), respondTo.expiresAtNanos(), new RequestFailure(reason, failure));
      }
  
      static Message<RequestFailure> failureResponse(long id, long expiresAtNanos, RequestFailure reason)
      {
          return outWithParam(id, Verb.FAILURE_RSP, expiresAtNanos, reason, null, null);
      }
  
      public <V> Message<V> withPayload(V newPayload)
      {
          return new Message<>(header, newPayload);
      }
  
      Message<T> withCallBackOnFailure()
      {
          return new Message<>(header.withFlag(MessageFlag.CALL_BACK_ON_FAILURE), payload);
      }
  
      public Message<T> withForwardTo(ForwardingInfo peers)
      {
          return new Message<>(header.withParam(ParamType.FORWARD_TO, peers), payload);
      }
  
      public Message<T> withFrom(InetAddressAndPort from)
      {
          return new Message<>(header.withFrom(from), payload);
      }
  
      public Message<T> withFlag(MessageFlag flag)
      {
          return new Message<>(header.withFlag(flag), payload);
      }
  
      public Message<T> withEpoch(Epoch epoch)
      {
          return new Message<>(header.withEpoch(epoch), payload);
      }
  
      public Message<T> withParam(ParamType type, Object value)
      {
          return new Message<>(header.withParam(type, value), payload);
      }
  
      public Message<T> withParams(Map<ParamType, Object> values)
      {
          if (values == null || values.isEmpty())
              return this;
          return new Message<>(header.withParams(values), payload);
      }
  
      private static final EnumMap<ParamType, Object> NO_PARAMS = new EnumMap<>(ParamType.class);
  
      private static Map<ParamType, Object> buildParams(ParamType type, Object value)
      {
          Map<ParamType, Object> params = NO_PARAMS;
          if (Tracing.isTracing())
              params = Tracing.instance.addTraceHeaders(new EnumMap<>(ParamType.class));
  
          if (type != null)
          {
              if (params.isEmpty())
                  params = new EnumMap<>(ParamType.class);
              params.put(type, value);
          }
  
          return params;
      }
  
      private static Map<ParamType, Object> addParam(Map<ParamType, Object> params, ParamType type, Object value)
      {
          if (type == null)
              return params;
  
          params = new EnumMap<>(params);
          params.put(type, value);
          return params;
      }
  
      private static Map<ParamType, Object> addParams(Map<ParamType, Object> params, Map<ParamType, Object> values)
      {
          if (values == null || values.isEmpty())
              return params;
  
          params = new EnumMap<>(params);
          params.putAll(values);
          return params;
      }
  
      /*
       * id generation
       */
  
      private static final long NO_ID = 0L; // this is a valid ID for pre40 nodes
  
      private static final AtomicInteger nextId = new AtomicInteger(0);
  
      private static long nextId()
      {
          long id;
          do
          {
              id = nextId.incrementAndGet();
          }
          while (id == NO_ID);
  
          return id;
      }
  
      /**
       * WARNING: this is inaccurate for messages from pre40 nodes, which can use 0 as an id (but will do so rarely)
       */
      @VisibleForTesting
      boolean hasId()
      {
          return id() != NO_ID;
      }
  
      /** we preface every message with this number so the recipient can validate the sender is sane */
      static final int PROTOCOL_MAGIC = 0xCA552DFA;
  
      static void validateLegacyProtocolMagic(int magic) throws InvalidLegacyProtocolMagic
      {
          if (magic != PROTOCOL_MAGIC)
              throw new InvalidLegacyProtocolMagic(magic);
      }
  
      public static final class InvalidLegacyProtocolMagic extends IOException
      {
          public final int read;
          private InvalidLegacyProtocolMagic(int read)
          {
              super(String.format("Read %d, Expected %d", read, PROTOCOL_MAGIC));
              this.read = read;
          }
      }
  
      public String toString()
      {
          return "(from:" + from() + ", type:" + verb().stage + " verb:" + verb() + ')';
      }
  
      /**
       * Split into a separate object to allow partial message deserialization without wasting work and allocation
       * afterwards, if the entire message is necessary and available.
       */
      public static class Header implements ResponseContext
      {
          public final long id;
          public final Epoch epoch;
          public final Verb verb;
          public final InetAddressAndPort from;
          public final long createdAtNanos;
          public final long expiresAtNanos;
          private final int flags;
          private final Map<ParamType, Object> params;
  
          private Header(long id, Epoch epoch, Verb verb, InetAddressAndPort from, long createdAtNanos, long expiresAtNanos, int flags, Map<ParamType, Object> params)
          {
              this.id = id;
              this.epoch = epoch;
              this.verb = verb;
              this.from = from;
              this.expiresAtNanos = expiresAtNanos;
              this.createdAtNanos = createdAtNanos;
              this.flags = flags;
              this.params = params;
          }
  
          public boolean hasFlag(MessageFlag messageFlag)
          {
              return messageFlag.isIn(flags);
          }
  
          Header withFrom(InetAddressAndPort from)
          {
              return new Header(id, epoch, verb, from, createdAtNanos, expiresAtNanos, flags, params);
          }
  
          Header withFlag(MessageFlag flag)
          {
              return new Header(id, epoch, verb, from, createdAtNanos, expiresAtNanos, flag.addTo(flags), params);
          }
  
          Header withEpoch(Epoch epoch)
          {
              return new Header(id, epoch, verb, from, createdAtNanos, expiresAtNanos, flags, params);
          }
  
          Header withParam(ParamType type, Object value)
          {
              return new Header(id, epoch, verb, from, createdAtNanos, expiresAtNanos, flags, addParam(params, type, value));
          }
  
          Header withParams(Map<ParamType, Object> values)
          {
              return new Header(id, epoch, verb, from, createdAtNanos, expiresAtNanos, flags, addParams(params, values));
          }
  
          boolean callBackOnFailure()
          {
              return MessageFlag.CALL_BACK_ON_FAILURE.isIn(flags);
          }
  
          boolean trackRepairedData()
          {
              return MessageFlag.TRACK_REPAIRED_DATA.isIn(flags);
          }
  
          boolean trackWarnings()
          {
              return MessageFlag.TRACK_WARNINGS.isIn(flags);
          }
  
          boolean isFinal()
          {
              return !MessageFlag.NOT_FINAL.isIn(flags);
          }
  
          @Nullable
          ForwardingInfo forwardTo()
          {
              return (ForwardingInfo) params.get(ParamType.FORWARD_TO);
          }
  
          @Nullable
          InetAddressAndPort respondTo()
          {
              InetAddressAndPort respondTo = (InetAddressAndPort) params.get(ParamType.RESPOND_TO);
              if (respondTo == null) respondTo = from;
              return respondTo;
          }
  
          /** Sender of the message. */
          @Override
          public InetAddressAndPort from()
          {
              return from;
          }
  
          @Override
          public long id()
          {
              return id;
          }
  
          @Override
          public Verb verb()
          {
              return verb;
          }
  
          @Override
          public long expiresAtNanos()
          {
              return expiresAtNanos;
          }
  
          @Nullable
          public TimeUUID traceSession()
          {
              return (TimeUUID) params.get(ParamType.TRACE_SESSION);
          }
  
          @Nullable
          public TraceType traceType()
          {
              return (TraceType) params.getOrDefault(ParamType.TRACE_TYPE, TraceType.QUERY);
          }
  
          public Map<ParamType, Object> params()
          {
              return Collections.unmodifiableMap(params);
          }
  
          @Nullable
          public Map<String,byte[]> customParams()
          {
              return (Map<String,byte[]>) params.get(ParamType.CUSTOM_MAP);
          }
      }
  
      @SuppressWarnings("WeakerAccess")
      public static class Builder<T>
      {
          private Verb verb;
          private InetAddressAndPort from;
          private T payload;
          private int flags = 0;
          private final Map<ParamType, Object> params = new EnumMap<>(ParamType.class);
          private long createdAtNanos;
          private long expiresAtNanos;
          private long id;
          private Epoch epoch;
  
          private boolean hasId;
  
          private Builder()
          {
          }
  
          public Builder<T> from(InetAddressAndPort from)
          {
              this.from = from;
              return this;
          }
  
          public Builder<T> withPayload(T payload)
          {
              this.payload = payload;
              return this;
          }
  
          public Builder<T> withFlag(MessageFlag flag)
          {
              flags = flag.addTo(flags);
              return this;
          }
  
          public Builder<T> withFlags(int flags)
          {
              this.flags = flags;
              return this;
          }
  
          public Builder<T> withParam(ParamType type, Object value)
          {
              params.put(type, value);
              return this;
          }
  
          public Builder<T> withCustomParam(String name, byte[] value)
          {
              Map<String,byte[]> customParams  = (Map<String,byte[]>)
                      params.computeIfAbsent(ParamType.CUSTOM_MAP, (t) -> new HashMap<String,byte[]>());
  
              customParams.put(name, value);
              return this;
          }
  
          /**
           * A shortcut to add tracing params.
           * Effectively, it is the same as calling {@link #withParam(ParamType, Object)} with tracing params
           * If there is already tracing params, calling this method overrides any existing ones.
           */
          public Builder<T> withTracingParams()
          {
              if (Tracing.isTracing())
                  Tracing.instance.addTraceHeaders(params);
              return this;
          }
  
          public Builder<T> withoutParam(ParamType type)
          {
              params.remove(type);
              return this;
          }
  
          public Builder<T> withParams(Map<ParamType, Object> params)
          {
              this.params.putAll(params);
              return this;
          }
  
          public Builder<T> ofVerb(Verb verb)
          {
              this.verb = verb;
              if (expiresAtNanos == 0 && verb != null && createdAtNanos != 0)
                  expiresAtNanos = verb.expiresAtNanos(createdAtNanos);
              if (!this.verb.isResponse() && from == null) // default to sending from self if we're a request verb
                  from = getBroadcastAddressAndPort();
              return this;
          }
  
          public Builder<T> withCreatedAt(long createdAtNanos)
          {
              this.createdAtNanos = createdAtNanos;
              if (expiresAtNanos == 0 && verb != null)
                  expiresAtNanos = verb.expiresAtNanos(createdAtNanos);
              return this;
          }
  
          public Builder<T> withExpiresAt(long expiresAtNanos)
          {
              this.expiresAtNanos = expiresAtNanos;
              return this;
          }
  
          public Builder<T> withId(long id)
          {
              this.id = id;
              hasId = true;
              return this;
          }
  
          public Builder<T> withEpoch(Epoch epoch)
          {
              this.epoch = epoch;
              return this;
          }
  
          public Message<T> build()
          {
              if (verb == null)
                  throw new IllegalArgumentException();
              if (from == null)
                  throw new IllegalArgumentException();
              if (payload == null)
                  throw new IllegalArgumentException();
              if (epoch == null)
                  epoch = epochSupplier.get();
  
              return new Message<>(new Header(hasId ? id : nextId(), epoch, verb, from, createdAtNanos, expiresAtNanos, flags, params), payload);
          }
      }
  
      public static <T> Builder<T> builder(Message<T> message)
      {
          return new Builder<T>().from(message.from())
                                 .withId(message.id())
                                 .ofVerb(message.verb())
                                 .withCreatedAt(message.createdAtNanos())
                                 .withExpiresAt(message.expiresAtNanos())
                                 .withFlags(message.header.flags)
                                 .withParams(message.header.params)
                                 .withEpoch(message.header.epoch)
                                 .withPayload(message.payload);
      }
  
      public static <T> Builder<T> builder(Verb verb, T payload)
      {
          return new Builder<T>().ofVerb(verb)
                                 .withCreatedAt(approxTime.now())
                                 .withPayload(payload);
      }
  
      public static final Serializer serializer = new Serializer();
  
      /**
       * Each message contains a header with several fixed fields, an optional key-value params section, and then
       * the message payload itself. Below is a visualization of the layout.
       *
       *  The params are prefixed by the count of key-value pairs; this value is encoded as unsigned vint.
       *  An individual param has an unsvint id (more specifically, a {@link ParamType}), and a byte array value.
       *  The param value is prefixed with it's length, encoded as an unsigned vint, followed by by the value's bytes.
       *
       * Legacy Notes (see {@link Serializer#serialize(Message, DataOutputPlus, int)} for complete details):
       * - pre 4.0, the IP address was sent along in the header, before the verb. The IP address may be either IPv4 (4 bytes) or IPv6 (16 bytes)
       * - pre-4.0, the verb was encoded as a 4-byte integer; in 4.0 and up it is an unsigned vint
       * - pre-4.0, the payloadSize was encoded as a 4-byte integer; in 4.0 and up it is an unsigned vint
       * - pre-4.0, the count of param key-value pairs was encoded as a 4-byte integer; in 4.0 and up it is an unsigned vint
       * - pre-4.0, param names were encoded as strings; in 4.0 they are encoded as enum id vints
       * - pre-4.0, expiry time wasn't encoded at all; in 4.0 it's an unsigned vint
       * - pre-4.0, message id was an int; in 4.0 and up it's an unsigned vint
       * - pre-4.0, messages included PROTOCOL MAGIC BYTES; post-4.0, we rely on frame CRCs instead
       * - pre-4.0, messages would serialize boolean params as dummy ONE_BYTEs; post-4.0 we have a dedicated 'flags' vint
       *
       * <pre>
       * {@code
       *            1 1 1 1 1 2 2 2 2 2 3
       *  0 2 4 6 8 0 2 4 6 8 0 2 4 6 8 0
       * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
       * | Message ID (vint)             |
       * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
       * | Epoch (vint)                  |
       * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
       * | Creation timestamp (int)      |
       * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
       * | Expiry (vint)                 |
       * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
       * | Verb (vint)                   |
       * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
       * | Flags (vint)                  |
       * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
       * | Param count (vint)            |
       * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
       * |                               /
       * /           Params              /
       * /                               |
       * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
       * | Payload size (vint)           |
       * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
       * |                               /
       * /           Payload             /
       * /                               |
       * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
       * }
       * </pre>
       */
      public static final class Serializer
      {
          private static final int CREATION_TIME_SIZE = 4;
  
          private Serializer()
          {
          }
  
          public <T> void serialize(Message<T> message, DataOutputPlus out, int version) throws IOException
          {
              serializeHeader(message.header, out, version);
              out.writeUnsignedVInt32(message.payloadSize(version));
              message.verb().serializer().serialize(message.payload, out, version);
          }
  
          public <T> Message<T> deserialize(DataInputPlus in, InetAddressAndPort peer, int version) throws IOException
          {
              Header header = deserializeHeader(in, peer, version);
              skipUnsignedVInt(in); // payload size, not needed by payload deserializer
              T payload = (T) header.verb.serializer().deserialize(in, version);
              return new Message<>(header, payload);
          }
  
          /**
           * A partial variant of deserialize, taking in a previously deserialized {@link Header} as an argument.
           *
           * Skip deserializing the {@link Header} from the input stream in favour of using the provided header.
           */
          public <T> Message<T> deserialize(DataInputPlus in, Header header, int version) throws IOException
          {
              skipHeader(in, version);
              skipUnsignedVInt(in); // payload size, not needed by payload deserializer
              T payload = (T) header.verb.serializer().deserialize(in, version);
              return new Message<>(header, payload);
          }
  
          private <T> int serializedSize(Message<T> message, int version)
          {
              long size = 0;
              size += serializedHeader(message.header, version);
              int payloadSize = message.payloadSize(version);
              size += sizeofUnsignedVInt(payloadSize) + payloadSize;
              return Ints.checkedCast(size);
          }
  
          /**
           * Size of the next message in the stream. Returns -1 if there aren't sufficient bytes read yet to determine size.
           */
          int inferMessageSize(ByteBuffer buf, int readerIndex, int readerLimit, int version)
          {
              int index = readerIndex;
  
              int idSize = computeUnsignedVIntSize(buf, index, readerLimit);
              if (idSize < 0)
                  return -1; // not enough bytes to read id
              index += idSize;
  
              if (version >= VERSION_51)
              {
                  int epochSize = computeUnsignedVIntSize(buf, index, readerLimit);
                  if (epochSize < 0)
                      return -1; // not enough bytes to read epoch
                  index += epochSize;
              }
  
              index += CREATION_TIME_SIZE;
              if (index > readerLimit)
                  return -1;
  
              int expirationSize = computeUnsignedVIntSize(buf, index, readerLimit);
              if (expirationSize < 0)
                  return -1;
              index += expirationSize;
  
              int verbIdSize = computeUnsignedVIntSize(buf, index, readerLimit);
              if (verbIdSize < 0)
                  return -1;
              index += verbIdSize;
  
              int flagsSize = computeUnsignedVIntSize(buf, index, readerLimit);
              if (flagsSize < 0)
                  return -1;
              index += flagsSize;
  
              int paramsSize = extractParamsSize(buf, index, readerLimit);
              if (paramsSize < 0)
                  return -1;
              index += paramsSize;
  
              long payloadSize = getUnsignedVInt(buf, index, readerLimit);
              if (payloadSize < 0)
                  return -1;
              index += computeUnsignedVIntSize(payloadSize) + payloadSize;
  
              int size = index - readerIndex;
  
              if (size > DatabaseDescriptor.getInternodeMaxMessageSizeInBytes())
                  throw new OversizedMessageException(size);
              return size;
          }
  
          /**
           * Partially deserialize the message - by only extracting the header and leaving the payload alone.
           *
           * To get the rest of the message without repeating the work done here, use {@link #deserialize(DataInputPlus, Header, int)}
           * method.
           *
           * It's assumed that the provided buffer contains all the bytes necessary to deserialize the header fully.
           */
          Header extractHeader(ByteBuffer buf, InetAddressAndPort from, long currentTimeNanos, int version) throws IOException
          {
              MonotonicClockTranslation timeSnapshot = approxTime.translate();
  
              int index = buf.position();
  
              long id = getUnsignedVInt(buf, index);
              index += computeUnsignedVIntSize(id);
  
              Epoch epoch = Epoch.EMPTY;
              if (version >= VERSION_51)
              {
                  long epochl = getUnsignedVInt(buf, index);
                  index += computeUnsignedVIntSize(epochl);
                  epoch = Epoch.create(epochl);
              }
              int createdAtMillis = buf.getInt(index);
              index += sizeof(createdAtMillis);
  
              long expiresInMillis = getUnsignedVInt(buf, index);
              index += computeUnsignedVIntSize(expiresInMillis);
  
              Verb verb = Verb.fromId(getUnsignedVInt32(buf, index));
              index += computeUnsignedVIntSize(verb.id);
  
              int flags = getUnsignedVInt32(buf, index);
              index += computeUnsignedVIntSize(flags);
  
              Map<ParamType, Object> params = extractParams(buf, index, version);
  
              long createdAtNanos = calculateCreationTimeNanos(createdAtMillis, timeSnapshot, currentTimeNanos);
              long expiresAtNanos = getExpiresAtNanos(createdAtNanos, TimeUnit.MILLISECONDS.toNanos(expiresInMillis));
  
              return new Header(id, epoch, verb, from, createdAtNanos, expiresAtNanos, flags, params);
          }
  
          private static long getExpiresAtNanos(long createdAtNanos, long expirationPeriodNanos)
          {
              return createdAtNanos + expirationPeriodNanos;
          }
  
          private void serializeHeader(Header header, DataOutputPlus out, int version) throws IOException
          {
              out.writeUnsignedVInt(header.id);
              if (version >= VERSION_51)
                  Epoch.messageSerializer.serialize(header.epoch, out, version);
              // int cast cuts off the high-order half of the timestamp, which we can assume remains
              // the same between now and when the recipient reconstructs it.
              out.writeInt((int) approxTime.translate().toMillisSinceEpoch(header.createdAtNanos));
              out.writeUnsignedVInt(NANOSECONDS.toMillis(header.expiresAtNanos - header.createdAtNanos));
              out.writeUnsignedVInt32(header.verb.id);
              out.writeUnsignedVInt32(header.flags);
              serializeParams(header.params, out, version);
          }
  
          public Header deserializeHeader(DataInputPlus in, InetAddressAndPort peer, int version) throws IOException
          {
              long id = in.readUnsignedVInt();
              Epoch epoch = Epoch.EMPTY;
              if (version >= VERSION_51)
                  epoch = Epoch.messageSerializer.deserialize(in, version);
              long currentTimeNanos = approxTime.now();
              MonotonicClockTranslation timeSnapshot = approxTime.translate();
              long creationTimeNanos = calculateCreationTimeNanos(in.readInt(), timeSnapshot, currentTimeNanos);
              long expiresAtNanos = getExpiresAtNanos(creationTimeNanos, TimeUnit.MILLISECONDS.toNanos(in.readUnsignedVInt()));
              Verb verb = Verb.fromId(in.readUnsignedVInt32());
              int flags = in.readUnsignedVInt32();
              Map<ParamType, Object> params = deserializeParams(in, version);
  
              return new Header(id, epoch, verb, peer, creationTimeNanos, expiresAtNanos, flags, params);
          }
  
          private void skipHeader(DataInputPlus in, int version) throws IOException
          {
              skipUnsignedVInt(in); // id
              if (version >= VERSION_51)
                  skipUnsignedVInt(in); // epoch
              in.skipBytesFully(4); // createdAt
              skipUnsignedVInt(in); // expiresIn
              skipUnsignedVInt(in); // verb
              skipUnsignedVInt(in); // flags
              skipParams(in); // params
          }
  
          private int serializedHeader(Header header, int version)
          {
              long size = 0;
              size += sizeofUnsignedVInt(header.id);
              if (version >= VERSION_51)
                  size += sizeofUnsignedVInt(header.epoch.getEpoch());
              size += CREATION_TIME_SIZE;
              size += sizeofUnsignedVInt(NANOSECONDS.toMillis(header.expiresAtNanos - header.createdAtNanos));
              size += sizeofUnsignedVInt(header.verb.id);
              size += sizeofUnsignedVInt(header.flags);
              size += serializedParamsSize(header.params, version);
              return Ints.checkedCast(size);
          }
  
          /*
           * created at + cross-node
           */
  
          private static final long TIMESTAMP_WRAPAROUND_GRACE_PERIOD_START  = 0xFFFFFFFFL - MINUTES.toMillis(15L);
          private static final long TIMESTAMP_WRAPAROUND_GRACE_PERIOD_END    =               MINUTES.toMillis(15L);
  
          @VisibleForTesting
          static long calculateCreationTimeNanos(int messageTimestampMillis, MonotonicClockTranslation timeSnapshot, long currentTimeNanos)
          {
              // We do not trust external time source, so we override their value with current time
              if (!DatabaseDescriptor.hasCrossNodeTimeout())
                  return currentTimeNanos;
  
              long currentTimeMillis = timeSnapshot.toMillisSinceEpoch(currentTimeNanos);
              // Reconstruct the message construction time sent by the remote host (we sent only the lower 4 bytes, assuming the
              // higher 4 bytes wouldn't change between the sender and receiver)
              long highBits = currentTimeMillis & 0xFFFFFFFF00000000L;
  
              long sentLowBits = messageTimestampMillis & 0x00000000FFFFFFFFL;
              long currentLowBits =   currentTimeMillis & 0x00000000FFFFFFFFL;
  
              // if our sent bits occur within a grace period of a wrap around event,
              // and our current bits are no more than the same grace period after a wrap around event,
              // assume a wrap around has occurred, and deduct one highBit
              if (      sentLowBits > TIMESTAMP_WRAPAROUND_GRACE_PERIOD_START
                        && currentLowBits < TIMESTAMP_WRAPAROUND_GRACE_PERIOD_END)
              {
                  highBits -= 0x0000000100000000L;
              }
              // if the message timestamp wrapped, but we still haven't, add one highBit
              else if (sentLowBits < TIMESTAMP_WRAPAROUND_GRACE_PERIOD_END
                       && currentLowBits > TIMESTAMP_WRAPAROUND_GRACE_PERIOD_START)
              {
                  highBits += 0x0000000100000000L;
              }
  
              long sentTimeMillis = (highBits | sentLowBits);
  
              if (Math.abs(currentTimeMillis - sentTimeMillis) > MINUTES.toMillis(15))
              {
                  noSpam1m.warn("Bad timestamp {} generated, overriding with currentTimeMillis = {}", sentTimeMillis, currentTimeMillis);
                  sentTimeMillis = currentTimeMillis;
              }
  
              return timeSnapshot.fromMillisSinceEpoch(sentTimeMillis);
          }
  
          /*
           * param ser/deser
           */
  
          private void serializeParams(Map<ParamType, Object> params, DataOutputPlus out, int version) throws IOException
          {
              out.writeUnsignedVInt32(params.size());
  
              for (Map.Entry<ParamType, Object> kv : params.entrySet())
              {
                  ParamType type = kv.getKey();
                  out.writeUnsignedVInt32(type.id);
  
                  IVersionedSerializer serializer = type.serializer;
                  Object value = kv.getValue();
  
                  int length = Ints.checkedCast(serializer.serializedSize(value, version));
                  out.writeUnsignedVInt32(length);
  
                  serializer.serialize(value, out, version);
              }
          }
  
          private Map<ParamType, Object> deserializeParams(DataInputPlus in, int version) throws IOException
          {
              int count = in.readUnsignedVInt32();
  
              if (count == 0)
                  return NO_PARAMS;
  
              Map<ParamType, Object> params = new EnumMap<>(ParamType.class);
  
              for (int i = 0; i < count; i++)
              {
                  ParamType type = ParamType.lookUpById(in.readUnsignedVInt32());
  
                  int length = in.readUnsignedVInt32();
  
                  if (null != type)
                  {
                      params.put(type, type.serializer.deserialize(in, version));
                  }
                  else
                  {
                      in.skipBytesFully(length); // forward compatibiliy with minor version changes
                  }
              }
  
              return params;
          }
  
          private Map<ParamType, Object> extractParams(ByteBuffer buf, int readerIndex, int version) throws IOException
          {
              long count = getUnsignedVInt(buf, readerIndex);
  
              if (count == 0)
                  return NO_PARAMS;
  
              final int position = buf.position();
              buf.position(readerIndex);
  
              try (DataInputBuffer in = new DataInputBuffer(buf, false))
              {
                  return deserializeParams(in, version);
              }
              finally
              {
                  buf.position(position);
              }
          }
  
          private void skipParams(DataInputPlus in) throws IOException
          {
              int count = in.readUnsignedVInt32();
  
              for (int i = 0; i < count; i++)
              {
                  skipUnsignedVInt(in);
                  in.skipBytesFully(in.readUnsignedVInt32());
              }
          }
  
          private long serializedParamsSize(Map<ParamType, Object> params, int version)
          {
              long size = computeUnsignedVIntSize(params.size());
  
              for (Map.Entry<ParamType, Object> kv : params.entrySet())
              {
                  ParamType type = kv.getKey();
                  Object value = kv.getValue();
  
                  long valueLength = type.serializer.serializedSize(value, version);
  
                  size += sizeofUnsignedVInt(type.id) + sizeofUnsignedVInt(valueLength);
  
                  size += valueLength;
              }
  
              return size;
          }
  
          private int extractParamsSize(ByteBuffer buf, int readerIndex, int readerLimit)
          {
              int index = readerIndex;
  
              long paramsCount = getUnsignedVInt(buf, index, readerLimit);
              if (paramsCount < 0)
                  return -1;
              index += computeUnsignedVIntSize(paramsCount);
  
              for (int i = 0; i < paramsCount; i++)
              {
                  long type = getUnsignedVInt(buf, index, readerLimit);
                  if (type < 0)
                      return -1;
                  index += computeUnsignedVIntSize(type);
  
                  long length = getUnsignedVInt(buf, index, readerLimit);
                  if (length < 0)
                      return -1;
                  index += computeUnsignedVIntSize(length) + length;
              }
  
              return index - readerIndex;
          }
  
          private <T> int payloadSize(Message<T> message, int version)
          {
              long payloadSize = message.payload != null && message.payload != NoPayload.noPayload
                               ? message.getPayloadSerializer().serializedSize(message.payload, version)
                               : 0;
              return Ints.checkedCast(payloadSize);
          }
      }
  
      private IVersionedAsymmetricSerializer<T, ?> getPayloadSerializer()
      {
          return payloadSerializer;
      }
  
      private int serializedSize40;
      private int serializedSize50;
      private int serializedSize51;
  
      /**
       * Serialized size of the entire message, for the provided messaging version. Caches the calculated value.
       */
      public int serializedSize(int version)
      {
          switch (version)
          {
              case VERSION_40:
                  if (serializedSize40 == 0)
                      serializedSize40 = serializer.serializedSize(this, VERSION_40);
                  return serializedSize40;
              case VERSION_50:
                  if (serializedSize50 == 0)
                      serializedSize50 = serializer.serializedSize(this, VERSION_50);
                  return serializedSize50;
              case VERSION_51:
                  if (serializedSize51 == 0)
                      serializedSize51 = serializer.serializedSize(this, VERSION_51);
                  return serializedSize51;
              default:
                  throw new IllegalStateException("Unknown serialization version " + version);
          }
      }
  
      private int payloadSize40 = -1;
      private int payloadSize50 = -1;
      private int payloadSize51 = -1;
  
      private int payloadSize(int version)
      {
          switch (version)
          {
              case VERSION_40:
                  if (payloadSize40 < 0)
                      payloadSize40 = serializer.payloadSize(this, VERSION_40);
                  return payloadSize40;
              case VERSION_50:
                  if (payloadSize50 < 0)
                      payloadSize50 = serializer.payloadSize(this, VERSION_50);
                  return payloadSize50;
              case VERSION_51:
                  if (payloadSize51 < 0)
                      payloadSize51 = serializer.payloadSize(this, VERSION_51);
                  return payloadSize51;
  
              default:
                  throw new IllegalStateException("Unkown serialization version " + version);
          }
      }
  
      static class OversizedMessageException extends RuntimeException
      {
          OversizedMessageException(int size)
          {
              super("Message of size " + size + " bytes exceeds allowed maximum of " + DatabaseDescriptor.getInternodeMaxMessageSizeInBytes() + " bytes");
          }
      }
  }
  