Merge remote-tracking branch 'Tesco/master'

README.MD

@@ -43,10 +43,10 @@ pieces of technology - a messaging system and a database.
 ## Introducing mewbase
 
 Mewbase provides an engine that functions as a high performance streaming event store but also
- allows you to run persistent *functions* in the engine that listen to raw events and maintain multiple *views* of
+ allows you to run persistent *functions* in the engine that listen to raw events and maintain multiple *projections* of
  the event data.
 
-The views can then be queried in similar way to how you would with a document or graph database.
+The projections can then be queried in similar way to how you would with a document or graph database.
 
 So you can think of mewbase as combining the following:
 

pom.xml

@@ -14,7 +14,16 @@
             <artifactId>vertx-core</artifactId>
             <version>3.3.3</version>
         </dependency>
-
+        <dependency>
+            <groupId>io.vertx</groupId>
+            <artifactId>vertx-auth-common</artifactId>
+            <version>3.3.3</version>
+        </dependency>
+        <dependency>
+            <groupId>io.vertx</groupId>
+            <artifactId>vertx-auth-shiro</artifactId>
+            <version>3.3.3</version>
+        </dependency>
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-core</artifactId>
@@ -72,7 +81,6 @@
             <scope>test</scope>
         </dependency>
 
-
     </dependencies>
 
     <build>

src/example/java/com/tesco/mewbase/example/ShoppingBasketExample.java

@@ -27,20 +27,19 @@ public static void main(String[] args) {
     private void example() throws Exception {
 
         // Setup and start a server
-        ServerOptions options = new ServerOptions().setChannels(new String[]{"orders"});
+        ServerOptions options =
+                new ServerOptions().setChannels(new String[]{"orders"}).setBinders(new String[]{"baskets"});
         Server server = Server.newServer(options);
         server.start().get();
 
-        // Install a function that will respond to add_item events and increase/decrease the quantity of the item in the basket
-        server.installFunction(
-                "basket_add",                                           // function name
-                "orders",                                               // channel name
-                ev -> ev.getString("eventType").equals("add_item"),     // event filter
-                "baskets",                                              // binder name
-                ev -> ev.getString("basketID"),                         // document id selector; how to obtain the doc id from the event bson
-                (basket, del) ->                                        // function to run
-                        BsonPath.add(basket, del.event().getInteger("quantity"), "products", del.event().getString("productID"))
-        );
+        // Register a projection that will respond to add_item events and increase/decrease the quantity of the item in the basket
+        server.buildProjection("maintain_basket")                             // projection name
+                .projecting("orders")                                           // channel name
+                .filteredBy(ev -> ev.getString("eventType").equals("add_item")) // event filter
+                .onto("baskets")                                                // binder name
+                .identifiedBy(ev -> ev.getString("basketID"))                   // document id selector; how to obtain the doc id from the event bson
+                .as((basket, del) ->                                            // projection function
+                        BsonPath.add(basket, del.event().getInteger("quantity"), "products", del.event().getString("productID"))).register();
 
         // Create a client
         Client client = Client.newClient(new ClientOptions());
@@ -59,7 +58,6 @@ private void example() throws Exception {
         // Now get the basket
         BsonObject basket = client.findByID("baskets", "basket1111").get();
 
-
         System.out.println("Basket is: " + basket);
     }
 }

src/example/java/com/tesco/mewbase/example/pubsub/PubSubExample.java

@@ -37,7 +37,8 @@ private void example() throws Exception {
         // Subscribe to a channel
         SubDescriptor descriptor = new SubDescriptor().setChannel("orders");
         client.subscribe(descriptor, del -> {
-            System.out.println("Received event: " + del.event().getString("foo"));
+            System.
+                    out.println("Received event: " + del.event().getString("foo"));
         });
 
         // Publish to the channel

src/main/java/com/tesco/mewbase/auth/MewbaseAuthProvider.java

@@ -0,0 +1,32 @@
+package com.tesco.mewbase.auth;
+
+import com.tesco.mewbase.bson.BsonObject;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ *
+ * User-facing interface for authenticating users.
+ *
+ */
+public interface MewbaseAuthProvider {
+
+    /**
+     * Authenticate a user.
+     * <p>
+     * The first argument is a Bson object containing information for authenticating the user. What this actually contains
+     * depends on the specific implementation. In the case of a simple username/password based
+     * authentication it is likely to contain a JSON object with the following structure:
+     * <pre>
+     *   {
+     *     "username": "tim",
+     *     "password": "mypassword"
+     *   }
+     * </pre>
+     *
+     * @param authInfo the auth information
+     * @return a CompletableFuture containing a MewbaseUser where authorization operations can be performed
+     *         on it
+     */
+    CompletableFuture<MewbaseUser> authenticate(BsonObject authInfo);
+}

src/main/java/com/tesco/mewbase/auth/MewbaseUser.java

@@ -0,0 +1,8 @@
+package com.tesco.mewbase.auth;
+
+/**
+ *  Contains all authorization info in order to perform operations
+ *  in the Mewbase code.
+ */
+public interface MewbaseUser {
+}

src/main/java/com/tesco/mewbase/auth/impl/DummyUser.java

@@ -0,0 +1,6 @@
+package com.tesco.mewbase.auth.impl;
+
+import com.tesco.mewbase.auth.MewbaseUser;
+
+public class DummyUser implements MewbaseUser {
+}

src/main/java/com/tesco/mewbase/auth/impl/MewbaseVertxAuthProvider.java

@@ -0,0 +1,54 @@
+package com.tesco.mewbase.auth.impl;
+
+import com.tesco.mewbase.auth.MewbaseAuthProvider;
+import com.tesco.mewbase.auth.MewbaseUser;
+import com.tesco.mewbase.bson.BsonObject;
+import io.vertx.core.json.JsonObject;
+import io.vertx.ext.auth.AuthProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ *
+ * This class serves as an adapter for the {@link io.vertx.ext.auth.AuthProvider}
+ *
+ * It will container a {@link io.vertx.ext.auth.AuthProvider} attribute, which is then used
+ * in the authenticate method.
+ *
+ * For the authentication information, the attributes just need to be inserted in the BsonObject.
+ *
+ * The mapping to {@link io.vertx.core.json.JsonObject} is done inside the authenticate method.
+ *
+ */
+public class MewbaseVertxAuthProvider implements MewbaseAuthProvider {
+
+    private final static Logger log = LoggerFactory.getLogger(MewbaseVertxAuthProvider.class);
+
+    private AuthProvider authProvider;
+
+    public MewbaseVertxAuthProvider(AuthProvider authProvider) {
+        this.authProvider = authProvider;
+    }
+
+    @Override
+    public CompletableFuture<MewbaseUser> authenticate(BsonObject authInfo) {
+        CompletableFuture<MewbaseUser> cf = new CompletableFuture<>();
+
+        JsonObject jsonAuthInfo = new JsonObject();
+        authInfo.stream().forEach(entry -> {
+            jsonAuthInfo.put(entry.getKey(), entry.getValue());
+        });
+
+        authProvider.authenticate(jsonAuthInfo, vertxRes -> {
+            if (vertxRes.succeeded()) {
+                cf.complete(new VertxUser(vertxRes.result()));
+            } else {
+                cf.completeExceptionally(vertxRes.cause());
+            }
+        });
+
+        return cf;
+    }
+}

src/main/java/com/tesco/mewbase/auth/impl/NoAuthAuthProvider.java

@@ -0,0 +1,18 @@
+package com.tesco.mewbase.auth.impl;
+
+import com.tesco.mewbase.auth.MewbaseAuthProvider;
+import com.tesco.mewbase.auth.MewbaseUser;
+import com.tesco.mewbase.bson.BsonObject;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * The default auth provider implementation that does not enforce any authentication
+ */
+public class NoAuthAuthProvider implements MewbaseAuthProvider {
+
+    @Override
+    public CompletableFuture<MewbaseUser> authenticate(BsonObject authInfo) {
+        return CompletableFuture.completedFuture(new DummyUser());
+    }
+}

src/main/java/com/tesco/mewbase/auth/impl/VertxUser.java

@@ -0,0 +1,13 @@
+package com.tesco.mewbase.auth.impl;
+
+import com.tesco.mewbase.auth.MewbaseUser;
+import io.vertx.ext.auth.User;
+
+public class VertxUser implements MewbaseUser {
+
+    private User user;
+
+    public VertxUser(User user) {
+        this.user = user;
+    }
+}

src/main/java/com/tesco/mewbase/bson/Bson.java

@@ -23,9 +23,8 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializerProvider;
 import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.tesco.mewbase.client.MewException;
 import de.undercouch.bson4jackson.BsonFactory;
-import io.vertx.core.json.DecodeException;
-import io.vertx.core.json.EncodeException;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -56,20 +55,20 @@ public class Bson {
         mapper.registerModule(module);
     }
 
-    public static void encode(Object obj, OutputStream outputStream) throws EncodeException {
+    public static void encode(Object obj, OutputStream outputStream) {
         try {
             mapper.writeValue(outputStream, obj);
         } catch (Exception e) {
-            throw new EncodeException("Failed to encode as BSON: " + e.getMessage());
+            throw new MewException(e);
         }
     }
 
 
-    public static <T> T decodeValue(InputStream inputStream, Class<T> clazz) throws DecodeException {
+    public static <T> T decodeValue(InputStream inputStream, Class<T> clazz) {
         try {
             return mapper.readValue(inputStream, clazz);
         } catch (Exception e) {
-            throw new DecodeException("Failed to decode BSON:" + e.getMessage());
+            throw new MewException(e);
         }
     }
 

src/main/java/com/tesco/mewbase/bson/BsonObject.java

@@ -774,6 +774,8 @@ protected void encodeToString(StringBuilder sb) {
                 sb.append("\"");
                 sb.append(entry.getValue().toString());
                 sb.append("\"");
+            } else if (entry.getValue() == null) {
+                sb.append("<null>");
             } else {
                 sb.append(entry.getValue().toString());
             }

src/main/java/com/tesco/mewbase/client/Client.java

@@ -23,6 +23,13 @@ static Client newClient(Vertx vertx, ClientOptions options) {
         return factory.newClient(vertx, options);
     }
 
+    // Error codes
+
+    int ERR_AUTHENTICATION_FAILED = 1;
+    int ERR_NOT_AUTHORISED = 2;
+    int ERR_NO_SUCH_CHANNEL = 3;
+    int ERR_FAILED_TO_PERSIST = 4;
+
     ClientFactory factory = ServiceHelper.loadFactory(ClientFactory.class);
 
     CompletableFuture<BsonObject> findByID(String binderName, String id);

src/main/java/com/tesco/mewbase/client/ClientOptions.java

@@ -1,5 +1,6 @@
 package com.tesco.mewbase.client;
 
+import com.tesco.mewbase.bson.BsonObject;
 import io.vertx.core.net.NetClientOptions;
 
 /**
@@ -13,6 +14,7 @@ public class ClientOptions {
     private String host = DEFAULT_HOST;
     private int port = DEFAULT_PORT;
     private NetClientOptions netClientOptions = new NetClientOptions();
+    private BsonObject authInfo;
 
     public String getHost() {
         return host;
@@ -41,6 +43,16 @@ public ClientOptions setNetClientOptions(NetClientOptions netClientOptions) {
         return this;
     }
 
+
+    public BsonObject getAuthInfo() {
+        return authInfo;
+    }
+
+    public ClientOptions setAuthInfo(BsonObject authInfo) {
+        this.authInfo = authInfo;
+        return this;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;
@@ -69,4 +81,5 @@ public String toString() {
                 ", port=" + port +
                 '}';
     }
+
 }

src/main/java/com/tesco/mewbase/client/MewException.java

@@ -5,27 +5,31 @@
  */
 public class MewException extends RuntimeException {
 
-    private final String errorCode;
+    private final int errorCode;
 
     public MewException(Exception e) {
-        this(null, e, null);
+        this(null, e, 0);
     }
 
     public MewException(String message) {
-        this(message, (String)null);
+        this(message, 0);
     }
 
-    public MewException(String message, Throwable cause, String errorCode) {
+    public MewException(String message, Throwable cause, int errorCode) {
         super(message, cause);
         this.errorCode = errorCode;
     }
 
     public MewException(String message, Throwable cause) {
-        this(message, cause, null);
+        this(message, cause, 0);
     }
 
-    public MewException(String message, String errorCode) {
+    public MewException(String message, int errorCode) {
         super(message);
         this.errorCode = errorCode;
     }
+
+    public int getErrorCode() {
+        return errorCode;
+    }
 }

src/main/java/com/tesco/mewbase/client/Subscription.java

@@ -15,8 +15,16 @@ public interface Subscription {
 
     // TODO return CompletableFuture<Void> ?
 
+    /**
+     * Unsubscribe a durable subscription
+     */
     void unsubscribe();
 
+    /**
+     * Close the subscription
+     */
+    void close();
+
     // Blocking yuck!
 
     BsonObject receive(long timeout);

src/main/java/com/tesco/mewbase/client/impl/ClientDeliveryImpl.java

@@ -10,8 +10,8 @@
  */
 public class ClientDeliveryImpl extends DeliveryImpl implements ClientDelivery {
 
-    private final SubscriptionImpl sub;
-    private final int sizeBytes;
+    protected final SubscriptionImpl sub;
+    protected final int sizeBytes;
 
     public ClientDeliveryImpl(String channel, long timestamp, long sequenceNumber, BsonObject event, SubscriptionImpl sub, int sizeBytes) {
         super(channel, timestamp, sequenceNumber, event);
@@ -21,7 +21,7 @@ public ClientDeliveryImpl(String channel, long timestamp, long sequenceNumber, B
 
     @Override
     public void acknowledge() {
-        sub.acknowledge(sizeBytes);
+        sub.acknowledge(channelPos, sizeBytes);
     }
 
     @Override