class VisibilityFence {

  public static TransactionAware create(byte[] fenceId) {
    return new ReadFence(fenceId);
  }

}

class VisibilityFenceTest {

  private static Configuration conf;
  private static TransactionManager txManager;

  @Test
  public void testFence2() throws Exception {

    byte[] fenceId = "test_table".getBytes(Charsets.UTF_8);

    // Readers use fence to indicate that they are interested in changes to specific data
    // Reader 1
    TransactionAware readFence1 = VisibilityFence.create(fenceId);
    TransactionContext readTxContext1 = new TransactionContext(new InMemoryTxSystemClient(txManager), readFence1);
    readTxContext1.start();

    // Reader 2
    TransactionAware readFence2 = VisibilityFence.create(fenceId);
    TransactionContext readTxContext2 = new TransactionContext(new InMemoryTxSystemClient(txManager), readFence2);
    readTxContext2.start();

    // Reader 3
    TransactionAware readFence3 = VisibilityFence.create(fenceId);
    TransactionContext readTxContext3 = new TransactionContext(new InMemoryTxSystemClient(txManager), readFence3);
    readTxContext3.start();

    // Writer updates data here in a separate transaction (code not shown)
    // start tx
    // update
    // commit tx

    // Now writer needs to wait for readers 1, 2, and 3 to see the change, it uses write fence to do so
    TransactionAware writeFence = new WriteFence(fenceId);
    TransactionContext writeTxContext = new TransactionContext(new InMemoryTxSystemClient(txManager), writeFence);
    writeTxContext.start();

    // Reader 1 commits before writeFence is committed
    readTxContext1.finish();

    try {
      // writeFence will throw exception since Reader 1 committed without seeing changes
      writeTxContext.finish();
      Assert.fail("Expected transaction to fail");
    } catch (TransactionConflictException e) {
      // Expected
      writeTxContext.abort();
    }

    // Start over writeFence again
    writeTxContext.start();

    // Now, Reader 3 commits before writeFence
    // Note that Reader 3 does not conflict with Reader 1
    readTxContext3.finish();

    try {
      // writeFence will throw exception again since Reader 3 committed without seeing changes
      writeTxContext.finish();
      Assert.fail("Expected transaction to fail");
    } catch (TransactionConflictException e) {
      // Expected
      writeTxContext.abort();
    }

    // Start over writeFence again
    writeTxContext.start();
    // This time writeFence commits before the other readers
    writeTxContext.finish();

    // After this point all readers will see the change

    try {
      // Reader 2 commits after writeFence, hence this commit with throw exception
      readTxContext2.finish();
      Assert.fail("Expected transaction to fail");
    } catch (TransactionConflictException e) {
      // Expected
      readTxContext2.abort();
    }

    // Reader 2 has to abort and start over again. It will see the changes now.
    readTxContext2 = new TransactionContext(new InMemoryTxSystemClient(txManager), readFence2);
    readTxContext2.start();
    readTxContext2.finish();
  }

}