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 accord.primitives;
  
  import java.util.Arrays;
  import java.util.Iterator;
  import java.util.Objects;
  import java.util.function.Consumer;
  import java.util.function.Function;
  import java.util.function.IntFunction;
  import java.util.function.Predicate;
  import java.util.stream.Stream;
  
  import javax.annotation.Nullable;
  
  import accord.api.RoutingKey;
  import accord.utils.ArrayBuffers.ObjectBuffers;
  import accord.utils.AsymmetricComparator;
  import accord.utils.IndexedBiConsumer;
  import accord.utils.IndexedFold;
  import accord.utils.IndexedFoldToLong;
  import accord.utils.IndexedTriFold;
  import accord.utils.Invariants;
  import accord.utils.SortedArrays;
  import net.nicoulaj.compilecommand.annotations.Inline;
  
  import static accord.primitives.Routable.Domain.Key;
  
  @SuppressWarnings("rawtypes")
  // TODO (desired, efficiency): check that foldl call-sites are inlined and optimised by HotSpot
  public abstract class AbstractKeys<K extends RoutableKey> implements Iterable<K>, Routables<K>
  {
      final K[] keys;
  
      protected AbstractKeys(K[] keys)
      {
          this.keys = keys;
      }
  
      @Override
      public boolean equals(Object o)
      {
          if (this == o) return true;
          if (o == null || getClass() != o.getClass()) return false;
          AbstractKeys that = (AbstractKeys) o;
          return Arrays.equals(keys, that.keys);
      }
  
      public final int indexOf(K key)
      {
          if (keys.length == 1)
          {
              int c = key.compareTo(keys[0]);
              if (c == 0) return 0;
              else if (c < 0) return -1;
              else return -2;
          }
          return Arrays.binarySearch(keys, key);
      }
  
      @Override
      public final K get(int indexOf)
      {
          return keys[indexOf];
      }
  
      @Override
      public final Routable.Domain domain()
      {
          return Key;
      }
  
      @Override
      public final boolean isEmpty()
      {
          return keys.length == 0;
      }
  
      @Override
      public final int size()
      {
          return keys.length;
      }
  
      @Override
      public final boolean intersects(AbstractUnseekableKeys keys)
      {
          return findNextIntersection(0, keys, 0) >= 0;
      }
  
      @Override
      public final boolean intersects(AbstractRanges ranges)
      {
          return findNextIntersection(0, ranges, 0) >= 0;
      }
  
      @Override
      public final int find(Range find, SortedArrays.Search search)
      {
          return SortedArrays.binarySearch(keys, 0, size(), find, Range::compareTo, search);
      }
  
      @Override
      public final int findNext(int thisIndex, Range find, SortedArrays.Search search)
      {
          return SortedArrays.exponentialSearch(keys, thisIndex, size(), find, Range::compareTo, search);
      }
  
      @Override
      public final long findNextIntersection(int thisIdx, AbstractRanges that, int thatIdx)
      {
          return SortedArrays.findNextIntersectionWithMultipleMatches(this.keys, thisIdx, that.ranges, thatIdx, (RoutableKey k, Range r) -> -r.compareTo(k), Range::compareTo);
      }
  
      @Override
      public final long findNextIntersection(int thisIdx, AbstractUnseekableKeys that, int thatIdx)
      {
          return SortedArrays.findNextIntersection(this.keys, thisIdx, that.keys, thatIdx, RoutableKey::compareAsRoutingKey);
      }
  
      @Override
      public final long findNextSameKindIntersection(int thisIdx, Routables<K> that, int thatIdx)
      {
          return SortedArrays.findNextIntersection(this.keys, thisIdx, ((AbstractKeys<?>)that).keys, thatIdx, RoutableKey::compareTo);
      }
  
      final boolean containsAllSameKind(AbstractKeys<?> that)
      {
          return that.size() == SortedArrays.foldlIntersection(keys, 0, keys.length, that.keys, 0, that.keys.length, (k, p, v, l, r) -> v + 1, 0, 0, 0);
      }
  
      @Override
      public Stream<K> stream()
      {
          return Stream.of(keys);
      }
  
      @Override
      public Iterator<K> iterator()
      {
          return new Iterator<>()
          {
              int i = 0;
              @Override
              public boolean hasNext()
              {
                  return i < keys.length;
              }
  
              @Override
              public K next()
              {
                  return keys[i++];
              }
          };
      }
  
      @Override
      public String toString()
      {
          return toString(null);
      }
  
      protected String toString(@Nullable RoutingKey printHomeKey)
      {
          if (isEmpty() && printHomeKey == null)
              return "[]";
  
          StringBuilder sb = new StringBuilder();
          sb.append('[');
          int i = 0;
          while (i < keys.length)
          {
              if (i > 0) sb.append(", ");
              Object prefix = keys[i].prefix();
              RoutingKey homeKey = null;
              if (printHomeKey != null && Objects.equals(prefix, printHomeKey.prefix()))
              {
                  homeKey = printHomeKey;
                  printHomeKey = null;
              }
              int j = i + 1;
              while (j < keys.length && Objects.equals(prefix, keys[j].prefix()))
                  ++j;
              if (prefix != null)
              {
                  sb.append(prefix);
                  sb.append(':');
                  sb.append('[');
              }
              while (i < j)
              {
                  K key = keys[i++];
                  sb.append(key.printableSuffix());
                  if (homeKey != null && Objects.equals(homeKey.suffix(), key.suffix()))
                  {
                      sb.append('*');
                      homeKey = null;
                  }
                  if (i < j) sb.append(',');
              }
              if (prefix != null)
              {
                  sb.append(']');
              }
              if (homeKey != null)
              {
                  sb.append('{');
                  sb.append(homeKey.printableSuffix());
                  sb.append("*}");
              }
          }
          if (printHomeKey == null) sb.append(']');
          else
          {
              if (keys.length > 0)
                  sb.append(", ");
  
              Object prefix = printHomeKey.prefix();
              if (prefix != null)
              {
                  sb.append(printHomeKey.prefix());
                  sb.append(":[");
              }
              sb.append("]{");
              sb.append(printHomeKey.printableSuffix());
              sb.append("*}");
              if (prefix != null)
                  sb.append(']');
          }
          return sb.toString();
      }
  
      // TODO (expected, efficiency): accept cached buffers
      protected K[] slice(AbstractRanges ranges, IntFunction<K[]> factory)
      {
          return SortedArrays.sliceWithMultipleMatches(keys, ranges.ranges, factory, (k, r) -> -r.compareTo(k), Range::compareTo);
      }
  
      protected static <K extends RoutableKey> K[] subtract(AbstractRanges ranges, K[] keys, IntFunction<K[]> factory)
      {
          return SortedArrays.subtractWithMultipleMatches(keys, ranges.ranges, factory, (k, r) -> -r.compareTo(k), Range::compareTo);
      }
  
      protected static <K extends RoutableKey> K[] subtract(Range range, K[] keys)
      {
          boolean isStartInclusive = range.startInclusive();
          int start = Arrays.binarySearch(keys, 0, keys.length, range.start(), RoutableKey::compareTo);
          if (start < 0) start = -1 - start;
          else if (!isStartInclusive) ++start;
          int end = Arrays.binarySearch(keys, 0, keys.length, range.end(), RoutableKey::compareTo);
          if (end < 0) end = -1 - end;
          else if (!isStartInclusive) ++end;
          if (start >= end)
              return Arrays.copyOf(keys, 0);
          return Arrays.copyOfRange(keys, start, end);
      }
  
      protected K[] intersecting(AbstractKeys<K> that, ObjectBuffers<K> buffers)
      {
          return SortedArrays.linearIntersection(this.keys, that.keys, buffers);
      }
  
      protected <K2 extends RoutableKey> K[] intersecting(AsymmetricComparator<K, K2> comparator, AbstractKeys<K2> that, ObjectBuffers<K> buffers)
      {
          return SortedArrays.asymmetricLinearIntersection(this.keys, this.keys.length, that.keys, that.keys.length, comparator, buffers);
      }
  
      protected K[] intersecting(AbstractRanges ranges, ObjectBuffers<K> buffers)
      {
          return SortedArrays.intersectWithMultipleMatches(keys, keys.length, ranges.ranges, ranges.ranges.length, (k, r) -> -r.compareTo(k), buffers);
      }
  
      public boolean any(Ranges ranges, Predicate<? super K> predicate)
      {
          return 1 == foldl(ranges, (key, p2, prev, index) -> predicate.test(key) ? 1 : 0, 0, 0, 1);
      }
  
      public boolean any(Predicate<? super K> predicate)
      {
          return 1 == foldl((key, p2, prev, index) -> predicate.test(key) ? 1 : 0, 0, 0, 1);
      }
  
      public boolean none(Predicate<? super K> predicate)
      {
          return !any(predicate);
      }
  
      @Inline
      public final <V> V foldl(Ranges rs, IndexedFold<? super K, V> fold, V accumulator)
      {
          return Routables.foldl(this, rs, fold, accumulator);
      }
  
      @Inline
      public final <P1, P2, V> V foldl(AbstractRanges intersect, IndexedTriFold<P1, P2, K, V> fold, P1 p1, P2 p2, V accumulator)
      {
          return Routables.foldl(this, intersect, fold, p1, p2, accumulator, i -> false);
      }
  
      @Inline
      public final void forEach(AbstractRanges rs, Consumer<? super K> forEach)
      {
          Routables.foldl(this, rs, (k, consumer, i) -> { consumer.accept(k); return consumer; }, forEach);
      }
  
      @Inline
      public final <P1> void forEach(AbstractRanges rs, IndexedBiConsumer<P1, ? super K> forEach, P1 p1)
      {
          Routables.foldl(this, rs, (p, ignore, k, consumer, i) -> {
              consumer.accept(p, k, i);
              return consumer;
          }, p1, null, forEach, i -> false);
      }
  
      @Inline
      public final <P1> void forEach(AbstractKeys<K> rs, IndexedBiConsumer<P1, ? super K> forEach, P1 p1)
      {
          Routables.foldl(this, rs, (p, ignore, k, consumer, i) -> {
              consumer.accept(p, k, i);
              return consumer;
          }, p1, null, forEach, i -> false);
      }
  
      @Inline
      public final long foldl(AbstractRanges rs, IndexedFoldToLong<? super K> fold, long param, long initialValue, long terminalValue)
      {
          return Routables.foldl(this, rs, fold, param, initialValue, terminalValue);
      }
  
      @Inline
      public final long foldl(AbstractKeys<K> that, IndexedFoldToLong<? super K> fold, long param, long initialValue, long terminalValue)
      {
          return Routables.foldl(this, that, fold, param, initialValue, terminalValue);
      }
  
      @Inline
      public final long foldl(IndexedFoldToLong<? super K> fold, long param, long initialValue, long terminalValue)
      {
          for (int i = 0; i < keys.length; i++)
          {
              initialValue = fold.apply(keys[i], param, initialValue, i);
              if (terminalValue == initialValue)
                  return initialValue;
          }
          return initialValue;
      }
  
      public void forEach(Consumer<? super K> forEach)
      {
          for (K key : keys)
              forEach.accept(key);
      }
  
      @Override
      public final FullKeyRoute toRoute(RoutingKey homeKey)
      {
          return toRoutingKeysArray(homeKey, false, (routingKeys, homeKeyIndex, isParticipatingHomeKey) -> new FullKeyRoute(routingKeys[homeKeyIndex], routingKeys));
      }
  
      protected RoutingKey[] toRoutingKeysArray(RoutingKey withKey, boolean permitInsert)
      {
          return toRoutingKeysArray(withKey, permitInsert, (routingKeys, homeKeyIndex, isParticipatingHomeKey) -> routingKeys);
      }
  
      interface ToRoutingKeysFactory<T>
      {
          T apply(RoutingKey[] keys, int insertPos, boolean includesKey);
      }
  
      protected <T> T toRoutingKeysArray(RoutingKey withKey, boolean permitInsert, ToRoutingKeysFactory<T> toRoutingKeysFactory)
      {
          RoutingKey[] copy;
          if (keys.getClass() == RoutingKey[].class)
          {
              copy = (RoutingKey[])keys;
          }
          else
          {
              copy = new RoutingKey[keys.length];
              int resultCount = copyToRoutingKeys(keys, 0, copy, 0, keys.length);
              if (resultCount < copy.length)
                  copy = Arrays.copyOf(copy, resultCount);
          }
  
          int insertPos = Arrays.binarySearch(copy, withKey);
          if (insertPos >= 0)
          {
              Invariants.require(copy[insertPos].equals(withKey));
              return toRoutingKeysFactory.apply(copy, insertPos, true);
          }
  
          if (!permitInsert)
              throw new IllegalArgumentException(withKey + " is expected to be a member of " + Arrays.toString(copy));
  
          insertPos = -1 - insertPos;
          RoutingKey[] newResult = new RoutingKey[1 + copy.length];
          System.arraycopy(copy, 0, newResult, 0, insertPos);
          newResult[insertPos] = withKey;
          System.arraycopy(copy, insertPos, newResult, insertPos + 1, copy.length - insertPos);
          return toRoutingKeysFactory.apply(newResult, insertPos, false);
      }
  
      public final RoutingKeys toParticipants()
      {
          if (getClass() == RoutingKeys.class)
              return (RoutingKeys) this;
          return toUnseekables(array -> array.length == 0 ? RoutingKeys.EMPTY : new RoutingKeys(array));
      }
  
      private <R> R toUnseekables(Function<RoutingKey[], R> constructor)
      {
          if (isEmpty())
              constructor.apply(RoutingKeys.EMPTY.keys);
  
          RoutingKey[] result = new RoutingKey[keys.length];
          int resultCount = copyToRoutingKeys(keys, 0, result, 0, keys.length);
          if (resultCount < result.length)
              result = Arrays.copyOf(result, resultCount);
          return constructor.apply(result);
      }
  
      private static <K extends RoutableKey> int copyToRoutingKeys(K[] src, int srcPos, RoutingKey[] trg, int trgPos, int count)
      {
          if (count == 0)
              return 0;
  
          int srcEnd = srcPos + count;
          int trgStart = trgPos;
          if (trgPos == 0)
              trg[trgPos++] = src[srcPos++].toUnseekable();
  
          while (srcPos < srcEnd)
          {
              RoutingKey next = src[srcPos++].toUnseekable();
              if (!next.equals(trg[trgPos - 1]))
                  trg[trgPos++] = next;
          }
  
          return trgPos - trgStart;
      }
  
      static <T extends RoutableKey> T[] sort(T[] keys)
      {
          Arrays.sort(keys);
          return keys;
      }
  
      @Override
      public int hashCode()
      {
          return Arrays.hashCode(keys);
      }
  
  }
  