commit to switch to my laptop

pom.xml

@@ -71,6 +71,8 @@
         <version.maven-rpm>2.1.4</version.maven-rpm>
         <version.maven-shade>2.4.2</version.maven-shade>
         <version.maven-surefire>2.19.1</version.maven-surefire>
+        <version.jawampa>0.4.2</version.jawampa>
+        <version.spring.boot>1.4.0.RELEASE</version.spring.boot>
     </properties>
     <dependencyManagement>
         <dependencies>
@@ -274,10 +276,19 @@
                 <version>${version.slf4j}</version>
             </dependency>
             <dependency>
-                <!-- Import dependency management from Spring Boot -->
+                <groupId>ws.wamp.jawampa</groupId>
+                <artifactId>jawampa-core</artifactId>
+                <version>${version.jawampa}</version>
+            </dependency>
+            <dependency>
+                <groupId>ws.wamp.jawampa</groupId>
+                <artifactId>jawampa-netty</artifactId>
+                <version>${version.jawampa}</version>
+            </dependency>
+            <dependency>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-dependencies</artifactId>
-                <version>1.4.0.RELEASE</version>
+                <version>${version.spring.boot}</version>
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>

server/pom.xml

@@ -56,6 +56,12 @@
             <classifier>${os.detected.classifier}</classifier>
             <!-- Eclipse doesn't like this, but it works on the cmd line -->
         </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-tcnative-boringssl-static</artifactId>
+            <version>${version.netty-tcnative}</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
@@ -152,6 +158,14 @@
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
         </dependency>
+        <dependency>
+            <groupId>ws.wamp.jawampa</groupId>
+            <artifactId>jawampa-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>ws.wamp.jawampa</groupId>
+            <artifactId>jawampa-netty</artifactId>
+        </dependency>
     </dependencies>
     <build>
         <extensions>

server/src/main/java/timely/Server.java

@@ -42,10 +42,12 @@ import java.net.InetSocketAddress;
 import java.nio.charset.StandardCharsets;
 import java.util.Date;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import rx.Observable;
 import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.context.ConfigurableApplicationContext;
 import timely.api.response.TimelyException;
@@ -83,6 +85,13 @@ import timely.netty.websocket.timeseries.WSSuggestRequestHandler;
 import timely.store.DataStore;
 import timely.store.DataStoreFactory;
 import timely.store.MetaCacheFactory;
+import timely.wamp.StateObserver;
+import timely.wamp.VersionObserver;
+import ws.wamp.jawampa.*;
+import ws.wamp.jawampa.connection.IWampConnectorProvider;
+import ws.wamp.jawampa.transport.netty.NettyWampClientConnectorProvider;
+import ws.wamp.jawampa.transport.netty.NettyWampConnectionConfig;
+import ws.wamp.jawampa.transport.netty.WampServerWebsocketHandler;
 
 public class Server {
 
@@ -93,6 +102,7 @@ public class Server {
     private static final String OS_NAME = "os.name";
     private static final String OS_VERSION = "os.version";
     private static final String WS_PATH = "/websocket";
+    private static final String WAMP_PATH = "/wamp";
 
     protected static final CountDownLatch LATCH = new CountDownLatch(1);
     static ConfigurableApplicationContext applicationContext;
@@ -109,6 +119,8 @@ public class Server {
     protected Channel wsChannelHandle = null;
     protected DataStore dataStore = null;
     protected volatile boolean shutdown = false;
+    private WampRouter wampRouter = null;
+    private WampClient wampClient;
 
     private static boolean useEpoll() {
 
@@ -225,6 +237,10 @@ public class Server {
         } catch (TimelyException e) {
             LOG.error("Error flushing to server during shutdown", e);
         }
+        LOG.info("Shutting down WAMP Router");
+        wampClient.close().toBlocking().last();
+        wampRouter.close().toBlocking().last(); // Blocks until router is fully
+                                                // shut down
         MetaCacheFactory.close();
         WebSocketRequestDecoder.close();
         if (applicationContext != null) {
@@ -305,6 +321,7 @@ public class Server {
 
         final int wsPort = config.getWebsocket().getPort();
         final ServerBootstrap wsServer = new ServerBootstrap();
+        setupWampRouter();
         wsServer.group(wsBossGroup, wsWorkerGroup);
         wsServer.channel(channelClass);
         wsServer.handler(new LoggingHandler());
@@ -316,6 +333,9 @@ public class Server {
         final String wsAddress = ((InetSocketAddress) wsChannelHandle.localAddress()).getAddress().getHostAddress();
 
         shutdownHook();
+
+        setupWampClient(wsAddress, wsPort);
+
         LOG.info(
                 "Server started. Listening on {}:{} for TCP traffic, {}:{} for HTTP traffic, and {}:{} for WebSocket traffic",
                 putAddress, putPort, queryAddress, queryPort, wsAddress, wsPort);
@@ -431,6 +451,7 @@ public class Server {
             @Override
             protected void initChannel(SocketChannel ch) throws Exception {
                 ch.pipeline().addLast("ssl", sslCtx.newHandler(ch.alloc()));
+                ch.pipeline().addLast("wamp", new WampServerWebsocketHandler(WAMP_PATH, wampRouter));
                 ch.pipeline().addLast("httpServer", new HttpServerCodec());
                 ch.pipeline().addLast("aggregator", new HttpObjectAggregator(8192));
                 ch.pipeline().addLast("sessionExtractor", new WebSocketHttpCookieHandler(config));
@@ -454,4 +475,36 @@ public class Server {
 
     }
 
+    protected void setupWampRouter() throws ApplicationError {
+        wampRouter = new WampRouterBuilder().addRealm("default").build();
+    }
+
+    protected void setupWampClient(String wsAddress, int wsPort) throws Exception {
+        IWampConnectorProvider connectorProvider = new NettyWampClientConnectorProvider();
+        NettyWampConnectionConfig connectionConfiguration = new NettyWampConnectionConfig.Builder()
+                .withSslContext(SslContextBuilder.forClient().build())
+                .build();
+
+        // Create a builder and configure the client
+        WampClientBuilder builder = new WampClientBuilder();
+        String wampUri = String.format("wss://0.0.0.0:%d%s", wsPort, WAMP_PATH);
+        System.out.println(wampUri);
+        builder.withConnectorProvider(connectorProvider)
+                .withConnectionConfiguration(connectionConfiguration)
+                .withUri(wampUri)
+                .withRealm("timely")
+                .withInfiniteReconnects()
+                .withReconnectInterval(5, TimeUnit.SECONDS);
+        // Create a client through the builder. This will not immediately start
+        // a connection attempt
+        wampClient = builder.build();
+        Observable<WampClient.State> stateObservable = wampClient.statusChanged();
+        stateObservable.subscribe(new StateObserver());
+
+        wampClient.open();
+
+        Observable<Request> request = wampClient.registerProcedure("timely.version");
+        request.subscribe(new VersionObserver());
+
+    }
 }

server/src/main/java/timely/wamp/StateObserver.java

@@ -0,0 +1,29 @@
+package timely.wamp;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import rx.Observer;
+import ws.wamp.jawampa.WampClient;
+
+/**
+ * Timely Client state observer
+ */
+public class StateObserver implements Observer<WampClient.State> {
+
+    private static final Logger LOG = LoggerFactory.getLogger(StateObserver.class);
+
+    @Override
+    public void onCompleted() {
+        LOG.info("WAMP state completed");
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        LOG.error("Wamp State Error: {}", e);
+    }
+
+    @Override
+    public void onNext(WampClient.State state) {
+        LOG.info("WAMP state change: {}", state);
+    }
+}

server/src/main/java/timely/wamp/VersionObserver.java

@@ -0,0 +1,26 @@
+package timely.wamp;
+
+import rx.Observer;
+import timely.api.request.VersionRequest;
+import ws.wamp.jawampa.Request;
+
+/**
+ * Handles timely.version requests
+ */
+public class VersionObserver implements Observer<Request> {
+
+    @Override
+    public void onCompleted() {
+
+    }
+
+    @Override
+    public void onError(Throwable e) {
+
+    }
+
+    @Override
+    public void onNext(Request request) {
+        request.reply(VersionRequest.VERSION);
+    }
+}

server/src/test/java/timely/test/integration/websocket/WampIT.java

@@ -0,0 +1,110 @@
+package timely.test.integration.websocket;
+
+import io.netty.handler.ssl.SslContextBuilder;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.minicluster.MiniAccumuloCluster;
+import org.apache.accumulo.minicluster.MiniAccumuloConfig;
+import org.junit.*;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TemporaryFolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import rx.Observable;
+import timely.Configuration;
+import timely.TestServer;
+import timely.auth.AuthCache;
+import timely.test.IntegrationTest;
+import timely.test.TestConfiguration;
+import ws.wamp.jawampa.Reply;
+import ws.wamp.jawampa.WampClient;
+import ws.wamp.jawampa.WampClientBuilder;
+import ws.wamp.jawampa.connection.IWampConnectorProvider;
+import ws.wamp.jawampa.transport.netty.NettyWampClientConnectorProvider;
+import ws.wamp.jawampa.transport.netty.NettyWampConnectionConfig;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Integration tests for the operations available over WAMP
+ */
+@Category(IntegrationTest.class)
+public class WampIT {
+
+    private static final Logger LOG = LoggerFactory.getLogger(WampIT.class);
+    private static final Long TEST_TIME = System.currentTimeMillis();
+
+    @ClassRule
+    public static final TemporaryFolder temp = new TemporaryFolder();
+
+    private static MiniAccumuloCluster mac = null;
+    private static Configuration conf = null;
+    private static TestServer server = null;
+
+
+    @BeforeClass
+    public static void beforeClass() throws Exception {
+        final MiniAccumuloConfig macConfig = new MiniAccumuloConfig(temp.newFolder("mac"), "secret");
+        mac = new MiniAccumuloCluster(macConfig);
+        mac.start();
+        conf = TestConfiguration.createMinimalConfigurationForTest();
+        conf.getAccumulo().setInstanceName(mac.getInstanceName());
+        conf.getAccumulo().setZookeepers(mac.getZooKeepers());
+        conf.getSecurity().getSsl().setUseOpenssl(false);
+        conf.getSecurity().getSsl().setUseGeneratedKeypair(true);
+
+        server = new TestServer(conf);
+    }
+
+    @AfterClass
+    public static void afterClass() throws Exception {
+        server.shutdown();
+        mac.stop();
+    }
+
+    @Before
+    public void setup() throws Exception {
+        Connector con = mac.getConnector("root", "secret");
+        con.tableOperations().list().forEach(t -> {
+            if (t.startsWith("timely")) {
+                try {
+                    con.tableOperations().delete(t);
+                } catch (Exception e) {
+                }
+            }
+        });
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        AuthCache.resetSessionMaxAge();
+    }
+
+    @Test
+    public void testVersion() throws Exception {
+        //final TestServer m = new TestServer(conf);
+
+        final WampClient client;
+        IWampConnectorProvider connectorProvider = new NettyWampClientConnectorProvider();
+        NettyWampConnectionConfig connectionConfiguration = new NettyWampConnectionConfig.Builder()
+                .withSslContext(SslContextBuilder.forClient().build())
+                .build();
+
+        WampClientBuilder builder = new WampClientBuilder();
+        builder.withConnectorProvider(connectorProvider)
+                .withConnectionConfiguration(connectionConfiguration)
+                .withUri("wss://localhost:54323/wamp")
+                .withRealm("default")
+                .withInfiniteReconnects()
+                .withReconnectInterval(5, TimeUnit.SECONDS);
+        // Create a client through the builder. This will not immediately start
+        // a connection attempt
+        client = builder.build();
+        client.open();
+
+        Observable<Reply> oReply = client.call("timely.version");
+        Reply reply = oReply.toBlocking().last();
+        System.out.println(reply.toString());
+    }
+
+
+}