package com.fabahaba.redis.lua.locks;

import java.util.UUID;

import com.fabahaba.jedipus.client.RedisClient;
import com.fabahaba.jedipus.cluster.Node;
import com.fabahaba.jedipus.cluster.RedisClusterExecutor;
import com.fabahaba.jedipus.cluster.RedisClusterExecutor.ReadMode;
import com.fabahaba.jedipus.cmds.Cmd;
import com.fabahaba.jedipus.cmds.CmdByteArray;
import com.fabahaba.jedipus.cmds.RESP;
import com.fabahaba.jedipus.lua.LuaScript;
import com.fabahaba.redis.locks.AcquireReply;

public final class RedisMutex {

  private RedisMutex() {}

  public static final LuaScript TRY_ACQUIRE =
      LuaScript.fromResourcePath("/redis/locks/MUTEX.TRY_ACQUIRE.lua");

  public static final Cmd<AcquireReply> EVALSHA_ACQUIRE =
      Cmd.create("EVALSHA", AcquireReply.ACQUIRE_REPLY_ADAPTER);

  public static final Cmd<Object[]> EVALSHA_ACQUIRE_RAW = Cmd.createCast("EVALSHA");

  public static final LuaScript TRY_RELEASE =
      LuaScript.fromResourcePath("/redis/locks/MUTEX.TRY_RELEASE.lua");

  public static final Cmd<String> EVALSHA_RELEASE = Cmd.createStringReply("EVALSHA");
  public static final Cmd<byte[]> EVALSHA_RELEASE_RAW = Cmd.createCast("EVALSHA");

  public static void loadMissingScripts(final RedisClusterExecutor rce) {
    LuaScript.loadMissingScripts(rce, TRY_ACQUIRE, TRY_RELEASE);
  }

  public static void loadMissingScripts(final RedisClient client) {
    LuaScript.loadMissingScripts(client, TRY_ACQUIRE, TRY_RELEASE);
  }

  private static final byte[] NUM_KEYS = RESP.toBytes(1);

  public static CmdByteArray<AcquireReply> createDirectAcquireArgs(final String lockName,
      final String ownerId, final long pexpire) {

    return CmdByteArray.startBuilding(EVALSHA_ACQUIRE, 6).addArg(TRY_ACQUIRE.getSha1HexBytes())
        .addArg(NUM_KEYS).addSlotKey(lockName).addArg(ownerId).addArg(pexpire).create();
  }

  public static CmdByteArray<Object[]> createDirectAcquireArgs(final byte[] lockName,
      final byte[] ownerId, final byte[] pexpire) {

    return CmdByteArray.startBuilding(EVALSHA_ACQUIRE_RAW, 6).addArg(TRY_ACQUIRE.getSha1HexBytes())
        .addArg(NUM_KEYS).addSlotKey(lockName).addArg(ownerId).addArg(pexpire).create();
  }

  public static CmdByteArray<String> createDirectReleaseArgs(final String lockName,
      final String ownerId) {

    return CmdByteArray.startBuilding(EVALSHA_RELEASE, 5).addArg(TRY_RELEASE.getSha1HexBytes())
        .addArg(NUM_KEYS).addSlotKey(lockName).addArg(ownerId).create();
  }

  public static CmdByteArray<byte[]> createDirectReleaseArgs(final byte[] lockName,
      final byte[] ownerId) {

    return CmdByteArray.startBuilding(EVALSHA_RELEASE_RAW, 5).addArg(TRY_RELEASE.getSha1HexBytes())
        .addArg(NUM_KEYS).addSlotKey(lockName).addArg(ownerId).create();
  }

  public static void main(final String[] args) {
    readme();
  }

  static void readme() {
    final String lockName = "MY_LOCK";
    final String ownerId = UUID.randomUUID().toString();
    final long pexpire = 2000;

    final CmdByteArray<AcquireReply> acquireCmd =
        RedisMutex.createDirectAcquireArgs(lockName, ownerId, pexpire);

    final CmdByteArray<String> releaseCmd = RedisMutex.createDirectReleaseArgs(lockName, ownerId);

    try (final RedisClusterExecutor rce =
        RedisClusterExecutor.startBuilding(Node.create("localhost", 7000)).create()) {

      RedisMutex.loadMissingScripts(rce);

      for (;;) {
        final AcquireReply lockOwners = rce.apply(ReadMode.MASTER, acquireCmd.getSlot(),
            client -> RedisMutex.TRY_ACQUIRE.eval(client, acquireCmd), 1);

        if (lockOwners.getCurrentOwner().equals(ownerId)) {
          try {
            System.out.format("Lock '%s' acquired by '%s'.%n", lockName, ownerId);
            break;
          } finally {
            // Even if tryReleaseLock fails for any reason (network errors) the lock will expire
            // after pexpire has elapsed. null is returned if the lock was no longer owned.
            final String released = rce.apply(ReadMode.MASTER, acquireCmd.getSlot(),
                client -> RedisMutex.TRY_RELEASE.eval(client, releaseCmd), 0);

            if (released != null && released.equals(ownerId)) {
              System.out.format("Lock '%s' was released by '%s'.%n", lockName, ownerId);
            } else {
              System.out.format("Lock '%s' was no longer owned by '%s'.%n", lockName, ownerId);
            }
          }
        }

        System.out.format("Waiting for %dms before retrying.", lockOwners.getTTLMillis());
        Thread.sleep(lockOwners.getTTLMillis());
      }
    } catch (final InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new RuntimeException(e);
    }
  }
}

@@ -36,32 +36,8 @@
 s;%0A%0A
-import java.util.UUID;%0A%0A
 impo
@@ -120,137 +120,28 @@
 ter.
-Node;%0Aimport com.fabahaba.jedipus.cluster.RedisClusterExecutor;%0Aimport com.fabahaba.jedipus.cluster.RedisClusterExecutor.ReadMode
+RedisClusterExecutor
 ;%0Aim
@@ -2622,1950 +2622,7 @@
 %0A  %7D
-%0A%0A  public static void main(final String%5B%5D args) %7B%0A    readme();%0A  %7D%0A%0A  static void readme() %7B%0A    final String lockName = %22MY_LOCK%22;%0A    final String ownerId = UUID.randomUUID().toString();%0A    final long pexpire = 2000;%0A%0A    final CmdByteArray%3CAcquireReply%3E acquireCmd =%0A        RedisMutex.createDirectAcquireArgs(lockName, ownerId, pexpire);%0A%0A    final CmdByteArray%3CString%3E releaseCmd = RedisMutex.createDirectReleaseArgs(lockName, ownerId);%0A%0A    try (final RedisClusterExecutor rce =%0A        RedisClusterExecutor.startBuilding(Node.create(%22localhost%22, 7000)).create()) %7B%0A%0A      RedisMutex.loadMissingScripts(rce);%0A%0A      for (;;) %7B%0A        final AcquireReply lockOwners = rce.apply(ReadMode.MASTER, acquireCmd.getSlot(),%0A            client -%3E RedisMutex.TRY_ACQUIRE.eval(client, acquireCmd), 1);%0A%0A        if (lockOwners.getCurrentOwner().equals(ownerId)) %7B%0A          try %7B%0A            System.out.format(%22Lock '%25s' acquired by '%25s'.%25n%22, lockName, ownerId);%0A            break;%0A          %7D finally %7B%0A            // Even if tryReleaseLock fails for any reason (network errors) the lock will expire%0A            // after pexpire has elapsed. null is returned if the lock was no longer owned.%0A            final String released = rce.apply(ReadMode.MASTER, acquireCmd.getSlot(),%0A                client -%3E RedisMutex.TRY_RELEASE.eval(client, releaseCmd), 0);%0A%0A            if (released != null && released.equals(ownerId)) %7B%0A              System.out.format(%22Lock '%25s' was released by '%25s'.%25n%22, lockName, ownerId);%0A            %7D else %7B%0A              System.out.format(%22Lock '%25s' was no longer owned by '%25s'.%25n%22, lockName, ownerId);%0A            %7D%0A          %7D%0A        %7D%0A%0A        System.out.format(%22Waiting for %25dms before retrying.%22, lockOwners.getTTLMillis());%0A        Thread.sleep(lockOwners.getTTLMillis());%0A      %7D%0A    %7D catch (final InterruptedException e) %7B%0A      Thread.currentThread().interrupt();%0A      throw new RuntimeException(e);%0A    %7D%0A  %7D
 %0A%7D%0A

Add separate reademe pkg, for readme code snippets.