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.coordinate;
  
  import java.util.concurrent.ExecutionException;
  import java.util.function.BiConsumer;
  
  import org.junit.jupiter.api.Test;
  
  import accord.api.LocalListeners;
  import accord.api.Result;
  import accord.impl.mock.MockCluster;
  import accord.impl.mock.MockStore;
  import accord.local.Node;
  import accord.local.SafeCommand;
  import accord.local.SafeCommandStore;
  import accord.primitives.FullKeyRoute;
  import accord.primitives.FullRangeRoute;
  import accord.primitives.Keys;
  import accord.primitives.Ranges;
  import accord.primitives.Txn;
  import accord.primitives.TxnId;
  
  import static accord.Utils.id;
  import static accord.Utils.ids;
  import static accord.Utils.ranges;
  import static accord.Utils.writeTxn;
  import static accord.impl.IntKey.keys;
  import static accord.impl.IntKey.range;
  import static accord.primitives.Routable.Domain.Key;
  import static accord.primitives.Routable.Domain.Range;
  import static accord.primitives.Txn.Kind.Read;
  import static accord.primitives.Txn.Kind.Write;
  import static accord.utils.async.AsyncChainUtils.getUninterruptibly;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertNotNull;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  import static org.junit.jupiter.api.Assertions.fail;
  
  public class CoordinateTransactionTest
  {
      @Test
      void simpleTest() throws Throwable
      {
          try (MockCluster cluster = MockCluster.builder().build())
          {
              Node node = cluster.get(1);
              assertNotNull(node);
  
              TxnId txnId = node.nextTxnIdWithDefaultFlags(Write, Key);
              Keys keys = keys(10);
              Txn txn = writeTxn(keys);
              FullKeyRoute route = keys.toRoute(keys.get(0).toUnseekable());
              Result result = getUninterruptibly(CoordinateTransaction.coordinate(node, route, txnId, txn));
              assertEquals(MockStore.RESULT, result);
          }
      }
  
      @Test
      void simpleRangeTest() throws Throwable
      {
          try (MockCluster cluster = MockCluster.builder().build())
          {
              Node node = cluster.get(1);
              assertNotNull(node);
  
              TxnId txnId = node.nextTxnIdWithDefaultFlags(Read, Range);
              Ranges keys = ranges(range(1, 2));
              Txn txn = writeTxn(keys);
              FullRangeRoute route = keys.toRoute(keys.get(0).someIntersectingRoutingKey(null));
              Result result = getUninterruptibly(CoordinateTransaction.coordinate(node, route, txnId, txn));
              assertEquals(MockStore.RESULT, result);
          }
      }
  
      @Test
      void exclusiveSyncTest() throws Throwable
      {
          try (MockCluster cluster = MockCluster.builder().build())
          {
              Node node = cluster.get(1);
              assertNotNull(node);
  
              TxnId oldId1 = node.nextTxnIdWithDefaultFlags(Write, Key);
              TxnId oldId2 = node.nextTxnIdWithDefaultFlags(Write, Key);
  
              getUninterruptibly(CoordinateSyncPoint.exclusive(node, ranges(range(0, 1))));
              try
              {
                  Keys keys = keys(1);
                  Txn txn = writeTxn(keys);
                  FullKeyRoute route = keys.toRoute(keys.get(0).someIntersectingRoutingKey(null));
                  getUninterruptibly(CoordinateTransaction.coordinate(node, route, oldId1, txn));
                  fail();
              }
              catch (ExecutionException e)
              {
                  assertTrue(e.getCause() instanceof Invalidated || e.getCause() instanceof Preempted);
              }
  
              Keys keys = keys(2);
              Txn txn = writeTxn(keys);
              FullKeyRoute route = keys.toRoute(keys.get(0).someIntersectingRoutingKey(null));
              getUninterruptibly(CoordinateTransaction.coordinate(node, route, oldId2, txn));
          }
      }
  
      @Test
      void slowPathTest() throws Throwable
      {
          try (MockCluster cluster = MockCluster.builder().nodes(7).replication(7).build())
          {
              cluster.networkFilter.isolate(ids(5, 7));
  
              Node node = cluster.get(1);
              assertNotNull(node);
  
              Txn txn = writeTxn(keys(10));
              Result result = getUninterruptibly(cluster.get(id(1)).coordinate(txn));
              assertEquals(MockStore.RESULT, result);
          }
      }
  
      private TxnId coordinate(Node node, long clock, Keys keys) throws Throwable
      {
          TxnId txnId = node.nextTxnIdWithDefaultFlags(Write, Key);
          txnId = new TxnId(txnId.epoch(), txnId.hlc() + clock, 0, Write, Key, txnId.node);
          Txn txn = writeTxn(keys);
          Result result = getUninterruptibly(CoordinateTransaction.coordinate(node, node.computeRoute(txnId, txn.keys()), txnId, txn));
          assertEquals(MockStore.RESULT, result);
          return txnId;
      }
  
      @Test
      void multiKeyTest() throws Throwable
      {
          try (MockCluster cluster = MockCluster.builder().nodes(6).maxKey(600).build())
          {
              Node node = cluster.get(1);
              assertNotNull(node);
  
              TxnId txnId1 = coordinate(node, 100, keys(50, 350, 550));
              TxnId txnId2 = coordinate(node, 150, keys(250, 350, 450));
              TxnId txnId3 = coordinate(node, 125, keys(50, 60, 70, 80, 350, 550));
          }
      }
  
      @Test
      void writeOnlyTest() throws Throwable
      {
          try (MockCluster cluster = MockCluster.builder().nodes(3).replication(3).build())
          {
              cluster.networkFilter.isolate(ids(5, 7));
  
              Node node = cluster.get(1);
              assertNotNull(node);
  
              Keys keys = keys(10);
              Txn txn = new Txn.InMemory(keys, MockStore.read(Keys.EMPTY), MockStore.QUERY, MockStore.update(keys));
              Result result = getUninterruptibly(cluster.get(id(1)).coordinate(txn));
              assertEquals(MockStore.RESULT, result);
          }
      }
  
      @Test
      void readOnlyTest() throws Throwable
      {
          try (MockCluster cluster = MockCluster.builder().nodes(3).replication(3).build())
          {
              cluster.networkFilter.isolate(ids(5, 7));
  
              Node node = cluster.get(1);
              assertNotNull(node);
  
              Keys keys = keys(10);
              Txn txn = new Txn.InMemory(keys, MockStore.read(keys), MockStore.QUERY, MockStore.update(Keys.EMPTY));
              Result result = getUninterruptibly(cluster.get(id(1)).coordinate(txn));
              assertEquals(MockStore.RESULT, result);
          }
      }
  
      @Test
      void simpleTxnThenReadOnlyTest() throws Throwable
      {
          try (MockCluster cluster = MockCluster.builder().build())
          {
              Node node = cluster.get(1);
              assertNotNull(node);
  
              TxnId txnId = node.nextTxnIdWithDefaultFlags(Write, Key);
              Keys oneKey = keys(10);
              Keys twoKeys = keys(10, 20);
              Txn txn = new Txn.InMemory(oneKey, MockStore.read(oneKey), MockStore.QUERY, MockStore.update(twoKeys));
              Result result = getUninterruptibly(CoordinateTransaction.coordinate(node, txn.keys().toRoute(oneKey.get(0).toUnseekable()), txnId, txn));
              assertEquals(MockStore.RESULT, result);
  
              txn = new Txn.InMemory(oneKey, MockStore.read(oneKey), MockStore.QUERY, MockStore.update(Keys.EMPTY));
              result = getUninterruptibly(cluster.get(id(1)).coordinate(txn));
              assertEquals(MockStore.RESULT, result);
          }
      }
  
      private static LocalListeners.ComplexListener commandListener(BiConsumer<SafeCommandStore, SafeCommand> listener)
      {
          return (safeStore, command) -> {
              listener.accept(safeStore, command);
              return true;
          };
      }
  }
  