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.locator;
  
  import java.io.IOException;
  import java.util.Objects;
  import java.util.Set;
  
  import com.google.common.base.Preconditions;
  
  import org.apache.cassandra.db.TypeSizes;
  import org.apache.cassandra.dht.IPartitioner;
  import org.apache.cassandra.dht.IPartitionerDependentSerializer;
  import org.apache.cassandra.dht.Range;
  import org.apache.cassandra.dht.Token;
  import org.apache.cassandra.io.util.DataInputPlus;
  import org.apache.cassandra.io.util.DataOutputPlus;
  import org.apache.cassandra.utils.FBUtilities;
  
  import static org.apache.cassandra.dht.AbstractBounds.tokenSerializer;
  
  /**
   * A Replica represents an owning node for a copy of a portion of the token ring.
   *
   * It consists of:
   *  - the logical token range that is being replicated (i.e. for the first logical replica only, this will be equal
   *      to one of its owned portions of the token ring; all other replicas will have this token range also)
   *  - an endpoint (IP and port)
   *  - whether the range is replicated in full, or transiently (CASSANDRA-14404)
   *
   * In general, it is preferred to use a Replica to a Range&lt;Token&gt;, particularly when users of the concept depend on
   * knowledge of the full/transient status of the copy.
   *
   * That means you should avoid unwrapping and rewrapping these things and think hard about subtraction
   * and such and what the result is WRT to transientness. Definitely avoid creating fake Replicas with misinformation
   * about endpoints, ranges, or transientness.
   */
  public final class Replica implements Comparable<Replica>, Endpoint
  {
      public static final IPartitionerDependentSerializer<Replica> serializer = new Serializer();
  
      private final Range<Token> range;
      private final InetAddressAndPort endpoint;
      private final boolean full;
  
      public Replica(InetAddressAndPort endpoint, Range<Token> range, boolean full)
      {
          Preconditions.checkNotNull(endpoint);
          Preconditions.checkNotNull(range);
          this.endpoint = endpoint;
          this.range = range;
          this.full = full;
      }
  
      public Replica(InetAddressAndPort endpoint, Token start, Token end, boolean full)
      {
          this(endpoint, new Range<>(start, end), full);
      }
  
      public boolean equals(Object o)
      {
          if (this == o) return true;
          if (o == null || getClass() != o.getClass()) return false;
          Replica replica = (Replica) o;
          return full == replica.full &&
                 Objects.equals(endpoint, replica.endpoint) &&
                 Objects.equals(range, replica.range);
      }
  
      @Override
      public int compareTo(Replica o)
      {
          int c = range.compareTo(o.range);
          if (c == 0)
              c = endpoint.compareTo(o.endpoint);
          if (c == 0)
              c =  Boolean.compare(full, o.full);
          return c;
      }
  
      public int hashCode()
      {
          return Objects.hash(endpoint, range, full);
      }
  
      @Override
      public String toString()
      {
          return (full ? "Full" : "Transient") + '(' + endpoint() + ',' + range + ')';
      }
  
      @Override
      public final InetAddressAndPort endpoint()
      {
          return endpoint;
      }
  
      public boolean isSelf()
      {
          return endpoint.equals(FBUtilities.getBroadcastAddressAndPort());
      }
  
      public Range<Token> range()
      {
          return range;
      }
  
      public final boolean isFull()
      {
          return full;
      }
  
      public final boolean isTransient()
      {
          return !isFull();
      }
  
      /**
       * This is used exclusively in TokenMetadata to check if a portion of a range is already replicated
       * by an endpoint so that we only mark as pending the portion that is either not replicated sufficiently (transient
       * when we need full) or at all.
       *
       * If it's not replicated at all it needs to be pending because there is no data.
       * If it's replicated but only transiently and we need to replicate it fully it must be marked as pending until it
       * is available fully otherwise a read might treat this replica as full and not read from a full replica that has
       * the data.
       */
      public RangesAtEndpoint subtractSameReplication(RangesAtEndpoint toSubtract)
      {
          Set<Range<Token>> subtractedRanges = range().subtractAll(toSubtract.filter(r -> r.isFull() == isFull()).ranges());
          RangesAtEndpoint.Builder result = RangesAtEndpoint.builder(endpoint, subtractedRanges.size());
          for (Range<Token> range : subtractedRanges)
          {
              result.add(decorateSubrange(range));
          }
          return result.build();
      }
  
      /**
       * Don't use this method and ignore transient status unless you are explicitly handling it outside this method.
       *
       * This helper method is used by StorageService.calculateStreamAndFetchRanges to perform subtraction.
       * It ignores transient status because it's already being handled in calculateStreamAndFetchRanges.
       */
      public RangesAtEndpoint subtractIgnoreTransientStatus(Range<Token> subtract)
      {
          Set<Range<Token>> ranges = this.range.subtract(subtract);
          RangesAtEndpoint.Builder result = RangesAtEndpoint.builder(endpoint, ranges.size());
          for (Range<Token> subrange : ranges)
              result.add(decorateSubrange(subrange));
          return result.build();
      }
  
      public boolean contains(Range<Token> that)
      {
          return range().contains(that);
      }
  
      public boolean intersectsOnRange(Replica replica)
      {
          return range().intersects(replica.range());
      }
  
      public Replica decorateSubrange(Range<Token> subrange)
      {
          Preconditions.checkArgument(range.contains(subrange), range + " " + subrange);
          return new Replica(endpoint(), subrange, isFull());
      }
  
      public static Replica fullReplica(InetAddressAndPort endpoint, Range<Token> range)
      {
          return new Replica(endpoint, range, true);
      }
  
      public static Replica fullReplica(InetAddressAndPort endpoint, Token start, Token end)
      {
          return fullReplica(endpoint, new Range<>(start, end));
      }
  
      public static Replica transientReplica(InetAddressAndPort endpoint, Range<Token> range)
      {
          return new Replica(endpoint, range, false);
      }
  
      public static Replica transientReplica(InetAddressAndPort endpoint, Token start, Token end)
      {
          return transientReplica(endpoint, new Range<>(start, end));
      }
  
      public static class Serializer implements IPartitionerDependentSerializer<Replica>
      {
          @Override
          public void serialize(Replica t, DataOutputPlus out, int version) throws IOException
          {
              tokenSerializer.serialize(t.range, out, version);
              InetAddressAndPort.Serializer.inetAddressAndPortSerializer.serialize(t.endpoint, out, version);
              out.writeBoolean(t.isFull());
          }
  
          @Override
          public Replica deserialize(DataInputPlus in, IPartitioner partitioner, int version) throws IOException
          {
              Range<Token> range = (Range<Token>) tokenSerializer.deserialize(in, partitioner, version);
              InetAddressAndPort endpoint = InetAddressAndPort.Serializer.inetAddressAndPortSerializer.deserialize(in, version);
              boolean isFull = in.readBoolean();
              return new Replica(endpoint, range, isFull);
          }
  
          @Override
          public long serializedSize(Replica t, int version)
          {
              return tokenSerializer.serializedSize(t.range, version) +
                     InetAddressAndPort.Serializer.inetAddressAndPortSerializer.serializedSize(t.endpoint, version) +
                     TypeSizes.sizeof(t.isFull());
          }
      }
  }
  
  