/*
 * Copyright 2014 the original author or authors.
 *
 * Licensed 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 net.kuujo.copycat.lockservice;

import net.kuujo.copycat.Copycat;
import net.kuujo.copycat.cluster.ClusterConfig;
import net.kuujo.copycat.collections.AsyncLock;
import net.kuujo.copycat.protocol.VertxEventBusProtocol;
import net.kuujo.copycat.VertxExecutionContext;
import org.vertx.java.core.Future;
import org.vertx.java.core.Handler;
import org.vertx.java.core.eventbus.Message;
import org.vertx.java.core.json.JsonObject;
import org.vertx.java.platform.Verticle;

/**
 * This is a replicated lock service accessible over the Vert.x event bus.
 *
 * @author <a href="http://github.com/kuujo">Jordan Halterman</a>
 */
public class VertxEventBusLockService extends Verticle {
  private Copycat copycat;
  private AsyncLock lock;

  private final Handler<Message<JsonObject>> messageHandler = message -> {
    String action = message.body().getString("action");
    if (action == null) {
      message.reply(new JsonObject().putString("status", "error").putString("message", "No action specified"));
    } else {
      switch (action) {
        case "lock":
          lock.lock().whenComplete((result, error) -> {
            if (error == null) {
              message.reply(new JsonObject().putString("status", "ok"));
            } else {
              message.reply(new JsonObject().putString("status", "error").putString("message", error.getMessage()));
            }
          });
          break;
        case "unlock":
          lock.unlock().whenComplete((result, error) -> {
            if (error == null) {
              message.reply(new JsonObject().putString("status", "ok"));
            } else {
              message.reply(new JsonObject().putString("status", "error").putString("message", error.getMessage()));
            }
          });
          break;
        default:
          message.reply(new JsonObject().putString("status", "error").putString("message", "Invalid action " + action));
          break;
      }
    }
  };

  @Override
  public void start(final Future<Void> startResult) {
    String address = container.config().getString("address", "lock");

    // Create a Copycat cluster configuration using the Vert.x event bus protocol.
    ClusterConfig cluster = new ClusterConfig()
      .withProtocol(new VertxEventBusProtocol(vertx))
      .withLocalMember("eventbus://lock1")
      .withRemoteMembers("eventbus://lock2", "eventbus://lock3");

    // Create a Copycat instance and give it a reference to the local Vert.x instance in order to run
    // asynchronous calls on the Vert.x event loop.
    copycat = Copycat.create(cluster, new VertxExecutionContext(vertx));

    // Open the Copycat instance and create a lock instance.
    copycat.open().whenComplete((copycatResult, copycatError) -> {
      if (copycatError == null) {

        // Create and open the log.
        lock = copycat.getLock(address);
        lock.open().whenComplete((lockResult, lockError) -> {
          if (lockError == null) {

            // Register an event bus handler for operating on the lock.
            vertx.eventBus().registerHandler(address, messageHandler, result -> {
              if (result.succeeded()) {
                startResult.setResult(null);
              } else {
                startResult.setFailure(result.cause());
              }
            });

          } else {
            startResult.setFailure(lockError);
          }
        });

      } else {
        startResult.setFailure(copycatError);
      }
    });
  }

}

@@ -678,16 +678,64 @@
 opycat;%0A
+import net.kuujo.copycat.VertxExecutionContext;%0A
 import n
@@ -883,56 +883,8 @@
 ol;%0A
-import net.kuujo.copycat.VertxExecutionContext;%0A
 impo
@@ -3431,15 +3431,8 @@
     
- lock =
  cop
@@ -3452,17 +3452,48 @@
 address)
-;
+.whenComplete((lock, error) -%3E %7B
 %0A       
@@ -3489,24 +3489,26 @@
 %3E %7B%0A        
+  
 lock.open().
@@ -3555,24 +3555,26 @@
 %7B%0A          
+  
 if (lockErro
@@ -3599,16 +3599,18 @@
         
+ 
+ 
 // Regis
@@ -3661,16 +3661,18 @@
 e lock.%0A
+  
         
@@ -3741,24 +3741,26 @@
 result -%3E %7B%0A
+  
             
@@ -3795,32 +3795,34 @@
                 
+  
 startResult.setR
@@ -3840,32 +3840,34 @@
 ;%0A              
+  
 %7D else %7B%0A       
@@ -3851,32 +3851,34 @@
        %7D else %7B%0A
+  
                 
@@ -3923,34 +3923,38 @@
 ;%0A              
-%7D%0A
+  %7D%0A  
             %7D);%0A
@@ -3956,32 +3956,34 @@
  %7D);%0A%0A          
+  
 %7D else %7B%0A       
@@ -3967,32 +3967,34 @@
        %7D else %7B%0A
+  
             star
@@ -4026,34 +4026,38 @@
 ror);%0A          
-%7D%0A
+  %7D%0A  
         %7D);%0A%0A   
@@ -4048,24 +4048,35 @@
         %7D);%0A
+        %7D);
 %0A      %7D els

Update Vert.x lock example for asynchronous resource API.