You are viewing a plain text version of this content. The canonical link for it is here.
Posted to submarine-dev@hadoop.apache.org by li...@apache.org on 2019/10/22 04:22:52 UTC
[hadoop-submarine] branch master updated: SUBMARINE-248. Add
websocket interface to submarine workbench server.
This is an automated email from the ASF dual-hosted git repository.
liuxun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hadoop-submarine.git
The following commit(s) were added to refs/heads/master by this push:
new bc1c2b4 SUBMARINE-248. Add websocket interface to submarine workbench server.
bc1c2b4 is described below
commit bc1c2b48f825997632877bcd34ed22eaf78965b1
Author: Zac Zhou <zh...@apache.org>
AuthorDate: Mon Oct 14 16:05:51 2019 +0800
SUBMARINE-248. Add websocket interface to submarine workbench server.
### What is this PR for?
Add WebSocket interface to the submarine workbench server. So that the frontend and backend can have bidirectional communications.
### What type of PR is it?
Feature
### What is the Jira issue?
https://issues.apache.org/jira/browse/SUBMARINE-248
### How should this be tested?
https://travis-ci.org/yuanzac/hadoop-submarine/builds/599666968
### Questions:
* Does the licenses files need update? No
* Is there breaking changes for older versions? No
* Does this needs documentation? No
Author: Zac Zhou <zh...@apache.org>
Closes #56 from yuanzac/topic/SUBMARINE-248 and squashes the following commits:
5d3903a [Zac Zhou] add tests for websocket.
731c779 [Zac Zhou] Add websocket interface
---
pom.xml | 2 +
submarine-client/pom.xml | 4 +-
submarine-commons/commons-runtime/pom.xml | 3 +-
submarine-commons/commons-utils/pom.xml | 5 +-
.../commons/utils/SubmarineConfiguration.java | 7 +
submarine-commons/pom.xml | 3 +-
.../server-submitter/submitter-yarn/pom.xml | 4 +-
.../server-submitter/submitter-yarnservice/pom.xml | 6 +-
.../interpreter/python-interpreter/pom.xml | 2 +-
submarine-workbench/workbench-server/pom.xml | 20 +-
.../apache/submarine/server/WorkbenchServer.java | 39 ++-
.../java/org/apache/submarine/utils/TestUtils.java | 50 ++++
.../submarine/websocket/ConnectionManager.java | 102 +++++++
.../submarine/websocket/DateJsonDeserializer.java | 52 ++++
.../org/apache/submarine/websocket/Message.java | 82 ++++++
.../apache/submarine/websocket/NotebookServer.java | 142 ++++++++++
.../apache/submarine/websocket/NotebookSocket.java | 85 ++++++
.../websocket/NotebookSocketListener.java | 26 ++
.../websocket/NotebookWebSocketCreator.java | 39 +++
.../server/AbstractWorkbenchServerTest.java | 300 +++++++++++++++++++++
.../apache/submarine/server/JsonResponseTest.java | 27 +-
.../submarine/websocket/NotebookServerTest.java | 97 +++++++
22 files changed, 1068 insertions(+), 29 deletions(-)
diff --git a/pom.xml b/pom.xml
index 7a1c36f..3706e90 100644
--- a/pom.xml
+++ b/pom.xml
@@ -42,6 +42,7 @@
</licenses>
<properties>
+ <submarine.version>0.3.0-SNAPSHOT</submarine.version>
<!-- language versions -->
<java.version>1.8</java.version>
@@ -68,6 +69,7 @@
<jackson-databind.version>2.9.10</jackson-databind.version>
<jackson-annotations.version>2.9.10</jackson-annotations.version>
<commons-configuration.version>1.10</commons-configuration.version>
+ <commons-httpclient.version>3.1</commons-httpclient.version>
<cglib.version>2.2.2</cglib.version>
<mybatis.version>3.2.8</mybatis.version>
diff --git a/submarine-client/pom.xml b/submarine-client/pom.xml
index 64a6c30..fb3fd40 100644
--- a/submarine-client/pom.xml
+++ b/submarine-client/pom.xml
@@ -36,7 +36,7 @@
<dependency>
<groupId>org.apache.submarine</groupId>
<artifactId>commons-runtime</artifactId>
- <version>0.3.0-SNAPSHOT</version>
+ <version>${submarine.version}</version>
</dependency>
<dependency>
<groupId>org.yaml</groupId>
@@ -59,7 +59,7 @@
<dependency>
<groupId>org.apache.submarine</groupId>
<artifactId>commons-runtime</artifactId>
- <version>0.3.0-SNAPSHOT</version>
+ <version>${submarine.version}</version>
<type>test-jar</type>
<scope>test</scope>
</dependency>
diff --git a/submarine-commons/commons-runtime/pom.xml b/submarine-commons/commons-runtime/pom.xml
index 2aa13bd..6716cad 100644
--- a/submarine-commons/commons-runtime/pom.xml
+++ b/submarine-commons/commons-runtime/pom.xml
@@ -22,12 +22,13 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
- <artifactId>submarine-commons</artifactId>
<groupId>org.apache.submarine</groupId>
+ <artifactId>submarine-commons</artifactId>
<version>0.3.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
+ <groupId>org.apache.submarine</groupId>
<artifactId>commons-runtime</artifactId>
<version>0.3.0-SNAPSHOT</version>
<name>Submarine: Commons Runtime</name>
diff --git a/submarine-commons/commons-utils/pom.xml b/submarine-commons/commons-utils/pom.xml
index 2afdf59..e58d6ea 100644
--- a/submarine-commons/commons-utils/pom.xml
+++ b/submarine-commons/commons-utils/pom.xml
@@ -21,14 +21,15 @@
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
<parent>
- <artifactId>submarine-commons</artifactId>
<groupId>org.apache.submarine</groupId>
+ <artifactId>submarine-commons</artifactId>
<version>0.3.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
- <modelVersion>4.0.0</modelVersion>
+ <groupId>org.apache.submarine</groupId>
<artifactId>commons-utils</artifactId>
<version>0.3.0-SNAPSHOT</version>
<name>Submarine: Commons Utils</name>
diff --git a/submarine-commons/commons-utils/src/main/java/org/apache/submarine/commons/utils/SubmarineConfiguration.java b/submarine-commons/commons-utils/src/main/java/org/apache/submarine/commons/utils/SubmarineConfiguration.java
index cea3492..8fe12ea 100644
--- a/submarine-commons/commons-utils/src/main/java/org/apache/submarine/commons/utils/SubmarineConfiguration.java
+++ b/submarine-commons/commons-utils/src/main/java/org/apache/submarine/commons/utils/SubmarineConfiguration.java
@@ -267,6 +267,10 @@ public class SubmarineConfiguration extends XMLConfiguration {
return getInt(ConfVars.CLUSTER_HEARTBEAT_TIMEOUT);
}
+ public String getWebsocketMaxTextMessageSize() {
+ return getString(ConfVars.WORKBENCH_WEBSOCKET_MAX_TEXT_MESSAGE_SIZE);
+ }
+
private String getStringValue(String name, String d) {
String value = this.properties.get(name);
if (value != null) {
@@ -382,6 +386,7 @@ public class SubmarineConfiguration extends XMLConfiguration {
}
public enum ConfVars {
+ SUBMARINE_CONF_DIR("submarine.conf.dir", "conf"),
SERVER_ADDR("workbench.server.addr", "0.0.0.0"),
SERVER_PORT("workbench.server.port", 8080),
SERVER_SSL("workbench.server.ssl", false),
@@ -407,6 +412,8 @@ public class SubmarineConfiguration extends XMLConfiguration {
"failOverReadOnly=false&zeroDateTimeBehavior=convertToNull&useSSL=false"),
JDBC_USERNAME("jdbc.username", "submarine"),
JDBC_PASSWORD("jdbc.password", "password"),
+ WORKBENCH_WEBSOCKET_MAX_TEXT_MESSAGE_SIZE(
+ "workbench.websocket.max.text.message.size", "1024000"),
WORKBENCH_WEB_WAR("workbench.web.war", "submarine-workbench/workbench-web/dist");
private String varName;
diff --git a/submarine-commons/pom.xml b/submarine-commons/pom.xml
index fffcc1e..e306c3e 100644
--- a/submarine-commons/pom.xml
+++ b/submarine-commons/pom.xml
@@ -21,12 +21,13 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
- <artifactId>submarine</artifactId>
<groupId>org.apache.submarine</groupId>
+ <artifactId>submarine</artifactId>
<version>0.3.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
+ <groupId>org.apache.submarine</groupId>
<artifactId>submarine-commons</artifactId>
<version>0.3.0-SNAPSHOT</version>
<packaging>pom</packaging>
diff --git a/submarine-server/server-submitter/submitter-yarn/pom.xml b/submarine-server/server-submitter/submitter-yarn/pom.xml
index 2f45a21..f710afa 100644
--- a/submarine-server/server-submitter/submitter-yarn/pom.xml
+++ b/submarine-server/server-submitter/submitter-yarn/pom.xml
@@ -31,7 +31,7 @@
<dependency>
<groupId>org.apache.submarine</groupId>
<artifactId>submarine-client</artifactId>
- <version>0.3.0-SNAPSHOT</version>
+ <version>${submarine.version}</version>
<scope>compile</scope>
<optional>true</optional>
<exclusions>
@@ -202,7 +202,7 @@
<artifactId>commons-runtime</artifactId>
<type>test-jar</type>
<scope>test</scope>
- <version>0.3.0-SNAPSHOT</version>
+ <version>${submarine.version}</version>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
diff --git a/submarine-server/server-submitter/submitter-yarnservice/pom.xml b/submarine-server/server-submitter/submitter-yarnservice/pom.xml
index 57e9e82..fb0d155 100644
--- a/submarine-server/server-submitter/submitter-yarnservice/pom.xml
+++ b/submarine-server/server-submitter/submitter-yarnservice/pom.xml
@@ -142,19 +142,19 @@
<dependency>
<groupId>org.apache.submarine</groupId>
<artifactId>commons-runtime</artifactId>
- <version>0.3.0-SNAPSHOT</version>
+ <version>${submarine.version}</version>
<type>test-jar</type>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.submarine</groupId>
<artifactId>submarine-client</artifactId>
- <version>0.3.0-SNAPSHOT</version>
+ <version>${submarine.version}</version>
</dependency>
<dependency>
<groupId>org.apache.submarine</groupId>
<artifactId>commons-runtime</artifactId>
- <version>0.3.0-SNAPSHOT</version>
+ <version>${submarine.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
diff --git a/submarine-workbench/interpreter/python-interpreter/pom.xml b/submarine-workbench/interpreter/python-interpreter/pom.xml
index 33c19a0..2949c21 100644
--- a/submarine-workbench/interpreter/python-interpreter/pom.xml
+++ b/submarine-workbench/interpreter/python-interpreter/pom.xml
@@ -39,7 +39,7 @@
<dependency>
<groupId>org.apache.submarine</groupId>
<artifactId>interpreter-engine</artifactId>
- <version>0.3.0-SNAPSHOT</version>
+ <version>${submarine.version}</version>
</dependency>
<dependency>
<groupId>org.apache.zeppelin</groupId>
diff --git a/submarine-workbench/workbench-server/pom.xml b/submarine-workbench/workbench-server/pom.xml
index 8db9182..f1ee9f3 100644
--- a/submarine-workbench/workbench-server/pom.xml
+++ b/submarine-workbench/workbench-server/pom.xml
@@ -35,10 +35,11 @@
<name>Submarine: Workbench Server</name>
<dependencies>
+
<dependency>
<groupId>org.apache.submarine</groupId>
<artifactId>commons-utils</artifactId>
- <version>0.3.0-SNAPSHOT</version>
+ <version>${submarine.version}</version>
</dependency>
<dependency>
@@ -208,6 +209,23 @@
<artifactId>mybatis-generator-core</artifactId>
<version>${mybatis-generator.version}</version>
</dependency>
+ <dependency>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>commons-httpclient</groupId>
+ <artifactId>commons-httpclient</artifactId>
+ <version>${commons-httpclient.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>commons-logging</groupId>
+ <artifactId>commons-logging</artifactId>
+ </exclusion>
+ </exclusions>
+ <scope>test</scope>
+ </dependency>
</dependencies>
diff --git a/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/server/WorkbenchServer.java b/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/server/WorkbenchServer.java
index 72d3ce1..8e4a381 100644
--- a/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/server/WorkbenchServer.java
+++ b/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/server/WorkbenchServer.java
@@ -19,6 +19,7 @@
package org.apache.submarine.server;
import org.apache.log4j.PropertyConfigurator;
+import org.apache.submarine.websocket.NotebookServer;
import org.eclipse.jetty.http.HttpVersion;
import org.eclipse.jetty.server.HttpConfiguration;
import org.eclipse.jetty.server.HttpConnectionFactory;
@@ -29,11 +30,17 @@ import org.eclipse.jetty.server.SslConnectionFactory;
import org.eclipse.jetty.server.handler.ContextHandlerCollection;
import org.eclipse.jetty.server.session.SessionHandler;
import org.eclipse.jetty.servlet.DefaultServlet;
+import org.eclipse.jetty.servlet.ServletContextHandler;
import org.eclipse.jetty.servlet.ServletHolder;
import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.eclipse.jetty.util.thread.ThreadPool;
import org.eclipse.jetty.webapp.WebAppContext;
+import org.eclipse.jetty.websocket.servlet.WebSocketServlet;
+import org.glassfish.hk2.api.ServiceLocator;
+import org.glassfish.hk2.api.ServiceLocatorFactory;
+import org.glassfish.hk2.utilities.ServiceLocatorUtilities;
+import org.glassfish.hk2.utilities.binding.AbstractBinder;
import org.glassfish.jersey.server.ResourceConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -42,12 +49,14 @@ import org.apache.submarine.commons.utils.SubmarineConfiguration;
import org.apache.submarine.commons.utils.SubmarineConfiguration.ConfVars;
import javax.inject.Inject;
+import javax.inject.Singleton;
import java.io.File;
public class WorkbenchServer extends ResourceConfig {
private static final Logger LOG = LoggerFactory.getLogger(WorkbenchServer.class);
public static Server jettyWebServer;
+ public static ServiceLocator sharedServiceLocator;
private static SubmarineConfiguration conf = SubmarineConfiguration.create();
@@ -64,7 +73,23 @@ public class WorkbenchServer extends ResourceConfig {
// Web UI
final WebAppContext webApp = setupWebAppContext(contexts, conf);
+ // Add
+ sharedServiceLocator = ServiceLocatorFactory.getInstance().create("shared-locator");
+ ServiceLocatorUtilities.enableImmediateScope(sharedServiceLocator);
+ ServiceLocatorUtilities.bind(
+ sharedServiceLocator,
+ new AbstractBinder() {
+ @Override
+ protected void configure() {
+ bindAsContract(NotebookServer.class)
+ .to(WebSocketServlet.class)
+ .in(Singleton.class);
+ }
+ });
+
setupRestApiContextHandler(webApp, conf);
+ // Notebook server
+ setupNotebookServer(webApp, conf, sharedServiceLocator);
startServer();
}
@@ -113,7 +138,7 @@ public class WorkbenchServer extends ResourceConfig {
}
private static WebAppContext setupWebAppContext(ContextHandlerCollection contexts,
- SubmarineConfiguration conf) {
+ SubmarineConfiguration conf) {
WebAppContext webApp = new WebAppContext();
webApp.setContextPath("/");
File warPath = new File(conf.getString(ConfVars.WORKBENCH_WEB_WAR));
@@ -179,10 +204,20 @@ public class WorkbenchServer extends ResourceConfig {
}
server.addConnector(connector);
-
return server;
}
+ private static void setupNotebookServer(WebAppContext webapp,
+ SubmarineConfiguration conf, ServiceLocator serviceLocator) {
+ String maxTextMessageSize = conf.getWebsocketMaxTextMessageSize();
+ final ServletHolder servletHolder =
+ new ServletHolder(serviceLocator.getService(NotebookServer.class));
+ servletHolder.setInitParameter("maxTextMessageSize", maxTextMessageSize);
+
+ final ServletContextHandler context = new ServletContextHandler(ServletContextHandler.SESSIONS);
+ webapp.addServlet(servletHolder, "/ws/*");
+ }
+
private static SslContextFactory getSslContextFactory(SubmarineConfiguration conf) {
SslContextFactory sslContextFactory = new SslContextFactory();
diff --git a/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/utils/TestUtils.java b/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/utils/TestUtils.java
new file mode 100644
index 0000000..8142f15
--- /dev/null
+++ b/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/utils/TestUtils.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.submarine.utils;
+
+import org.apache.submarine.server.WorkbenchServer;
+import org.glassfish.hk2.api.ServiceLocator;
+import org.glassfish.hk2.api.ServiceLocatorFactory;
+
+import java.util.Arrays;
+
+public class TestUtils {
+ public static <T> T getInstance(Class<T> clazz) {
+ checkCalledByTestMethod();
+ return getInstance(WorkbenchServer.sharedServiceLocator, clazz);
+ }
+
+ public static void clearInstances() {
+ checkCalledByTestMethod();
+ ServiceLocatorFactory.getInstance().destroy("shared-locator");
+ }
+
+ static <T> T getInstance(ServiceLocator serviceLocator, Class<T> clazz) {
+ return serviceLocator.getService(clazz);
+ }
+
+ static void checkCalledByTestMethod() {
+ StackTraceElement[] stackTraceElements = Thread.currentThread().getStackTrace();
+ // The first element of [0] indicates 'java.lang.Thread.getStackTrace'.
+ // The second element of [1] indicates this method.
+ // The third element of [2] indicates a caller of this method.
+ if (Arrays.stream(stackTraceElements)
+ .noneMatch(stackTraceElement -> stackTraceElement.getClassName().contains("Test"))) {
+ throw new RuntimeException("This method shouldn't be used in production");
+ }
+ }
+}
diff --git a/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/ConnectionManager.java b/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/ConnectionManager.java
new file mode 100644
index 0000000..3ed92e7
--- /dev/null
+++ b/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/ConnectionManager.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.submarine.websocket;
+
+import com.google.common.collect.Sets;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import org.eclipse.jetty.websocket.api.WebSocketException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Date;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * Manager class for managing websocket connections
+ */
+public class ConnectionManager {
+ private static final Logger LOG = LoggerFactory.getLogger(ConnectionManager.class);
+ private static Gson gson = new GsonBuilder()
+ .setDateFormat("yyyy-MM-dd'T'HH:mm:ssZ")
+ .registerTypeAdapter(Date.class, new DateJsonDeserializer())
+ .setPrettyPrinting()
+ .create();
+
+ final Queue<NotebookSocket> connectedSockets = new ConcurrentLinkedQueue<>();
+ // user -> connection
+ final Map<String, Queue<NotebookSocket>> userSocketMap = new ConcurrentHashMap<>();
+
+ public void addConnection(NotebookSocket conn) {
+ connectedSockets.add(conn);
+ }
+
+ public void removeConnection(NotebookSocket conn) {
+ connectedSockets.remove(conn);
+ }
+
+ public void addUserConnection(String user, NotebookSocket conn) {
+ LOG.info("Add user connection {} for user: {}", conn, user);
+ conn.setUser(user);
+ if (userSocketMap.containsKey(user)) {
+ userSocketMap.get(user).add(conn);
+ } else {
+ Queue<NotebookSocket> socketQueue = new ConcurrentLinkedQueue<>();
+ socketQueue.add(conn);
+ userSocketMap.put(user, socketQueue);
+ }
+ }
+
+ public void removeUserConnection(String user, NotebookSocket conn) {
+ LOG.info("Remove user connection {} for user: {}", conn, user);
+ if (userSocketMap.containsKey(user)) {
+ userSocketMap.get(user).remove(conn);
+ } else {
+ LOG.warn("Closing connection that is absent in user connections");
+ }
+ }
+
+ protected String serializeMessage(Message m) {
+ return gson.toJson(m);
+ }
+
+ public void broadcast(Message m) {
+ synchronized (connectedSockets) {
+ for (NotebookSocket ns : connectedSockets) {
+ try {
+ ns.send(serializeMessage(m));
+ } catch (IOException | WebSocketException e) {
+ LOG.error("Send error: " + m, e);
+ }
+ }
+ }
+ }
+
+ public Set<String> getConnectedUsers() {
+ Set<String> connectedUsers = Sets.newHashSet();
+ for (NotebookSocket notebookSocket : connectedSockets) {
+ connectedUsers.add(notebookSocket.getUser());
+ }
+ return connectedUsers;
+ }
+}
diff --git a/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/DateJsonDeserializer.java b/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/DateJsonDeserializer.java
new file mode 100644
index 0000000..07885b1
--- /dev/null
+++ b/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/DateJsonDeserializer.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.submarine.websocket;
+
+import com.google.gson.JsonDeserializationContext;
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonParseException;
+
+import java.lang.reflect.Type;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.Locale;
+
+public class DateJsonDeserializer implements JsonDeserializer{
+ private final String[] DATE_FORMATS = new String[] {
+ "yyyy-MM-dd'T'HH:mm:ssZ",
+ "MMM d, yyyy h:mm:ss a",
+ "MMM dd, yyyy HH:mm:ss",
+ "yyyy-MM-dd HH:mm:ss.SSS"
+ };
+
+ @Override
+ public Date deserialize(JsonElement jsonElement, Type typeOF,
+ JsonDeserializationContext context) throws JsonParseException {
+ for (String format : DATE_FORMATS) {
+ try {
+ return new SimpleDateFormat(format, Locale.US).parse(jsonElement.getAsString());
+ } catch (ParseException e) {
+ // do nothing
+ }
+ }
+ throw new JsonParseException("Unparsable date: \"" + jsonElement.getAsString()
+ + "\". Supported formats: " + Arrays.toString(DATE_FORMATS));
+ }
+}
diff --git a/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/Message.java b/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/Message.java
new file mode 100644
index 0000000..93ade25
--- /dev/null
+++ b/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/Message.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.submarine.websocket;
+
+import com.google.gson.Gson;
+import org.slf4j.Logger;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class Message {
+ /**
+ * Representation of event type.
+ */
+ public enum OP {
+ ERROR_INFO, // [s-c] error information to be sent
+ NOTICE // [s-c] Notice
+ }
+
+ private static final Gson gson = new Gson();
+ public static final Message EMPTY = new Message(null);
+
+ public OP op;
+ public Map<String, Object> data = new HashMap<>();
+
+ public Message(OP op) {
+ this.op = op;
+ }
+
+ public Message put(String k, Object v) {
+ data.put(k, v);
+ return this;
+ }
+
+ public Object get(String k) {
+ return data.get(k);
+ }
+
+ public <T> T getType(String key) {
+ return (T) data.get(key);
+ }
+
+ public <T> T getType(String key, Logger LOG) {
+ try {
+ return getType(key);
+ } catch (ClassCastException e) {
+ LOG.error("Failed to get " + key + " from message (Invalid type). " , e);
+ return null;
+ }
+ }
+
+ @Override
+ public String toString() {
+ final StringBuilder sb = new StringBuilder("Message{");
+ sb.append("data=").append(data);
+ sb.append(", op=").append(op);
+ sb.append('}');
+ return sb.toString();
+ }
+
+ public String toJson() {
+ return gson.toJson(this);
+ }
+
+ public static Message fromJson(String json) {
+ return gson.fromJson(json, Message.class);
+ }
+}
diff --git a/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/NotebookServer.java b/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/NotebookServer.java
new file mode 100644
index 0000000..6a42a01
--- /dev/null
+++ b/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/NotebookServer.java
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.submarine.websocket;
+
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import java.io.IOException;
+import java.util.Date;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.commons.lang.StringUtils;
+import org.eclipse.jetty.util.annotation.ManagedAttribute;
+import org.eclipse.jetty.util.annotation.ManagedObject;
+import org.eclipse.jetty.util.annotation.ManagedOperation;
+import org.eclipse.jetty.websocket.servlet.WebSocketServlet;
+import org.eclipse.jetty.websocket.servlet.WebSocketServletFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Submarine websocket service. This class used setter injection because all servlet should have
+ * no-parameter constructor
+ */
+@ManagedObject
+public class NotebookServer extends WebSocketServlet
+ implements NotebookSocketListener {
+
+ /**
+ * Job manager service type.
+ */
+ protected enum JobManagerServiceType {
+ JOB_MANAGER_PAGE("JOB_MANAGER_PAGE");
+ private String serviceTypeKey;
+
+ JobManagerServiceType(String serviceType) {
+ this.serviceTypeKey = serviceType;
+ }
+
+ String getKey() {
+ return this.serviceTypeKey;
+ }
+ }
+
+ private static final Logger LOG = LoggerFactory.getLogger(NotebookServer.class);
+ private static Gson gson = new GsonBuilder()
+ .setDateFormat("yyyy-MM-dd'T'HH:mm:ssZ")
+ .registerTypeAdapter(Date.class, new DateJsonDeserializer())
+ .setPrettyPrinting()
+ .create();
+
+ private static AtomicReference<NotebookServer> self = new AtomicReference<>();
+
+ private ConnectionManager connectionManager;
+
+ private ExecutorService executorService = Executors.newFixedThreadPool(10);
+
+ public NotebookServer() {
+ this.connectionManager = new ConnectionManager();
+ NotebookServer.self.set(this);
+ LOG.info("NotebookServer instantiated: {}", this);
+ }
+
+ @Override
+ public void configure(WebSocketServletFactory factory) {
+ factory.setCreator(new NotebookWebSocketCreator(this));
+ }
+
+ @Override
+ public void onOpen(NotebookSocket conn) {
+ LOG.info("New connection from {}", conn);
+ connectionManager.addConnection(conn);
+ }
+
+ @Override
+ public void onMessage(NotebookSocket conn, String msg) {
+ try {
+ LOG.info("Got Message: " + msg);
+ if (StringUtils.isEmpty(conn.getUser())) {
+ connectionManager.addUserConnection("FakeUser1", conn);
+ }
+ } catch (Exception e) {
+ LOG.error("Can't handle message: " + msg, e);
+ try {
+ conn.send(serializeMessage(new Message(Message.OP.ERROR_INFO).put(
+ "info", e.getMessage())));
+ } catch (IOException iox) {
+ LOG.error("Fail to send error info", iox);
+ }
+ }
+ }
+
+ @Override
+ public void onClose(NotebookSocket conn, int code, String reason) {
+ LOG.info("Closed connection to {} ({}) {}", conn, code, reason);
+ connectionManager.removeConnection(conn);
+ connectionManager.removeUserConnection(conn.getUser(), conn);
+ }
+
+ public ConnectionManager getConnectionManager() {
+ return connectionManager;
+ }
+
+ protected Message deserializeMessage(String msg) {
+ return gson.fromJson(msg, Message.class);
+ }
+
+ protected String serializeMessage(Message m) {
+ return gson.toJson(m);
+ }
+
+ public void broadcast(Message m) {
+ connectionManager.broadcast(m);
+ }
+
+ @ManagedAttribute
+ public Set<String> getConnectedUsers() {
+ return connectionManager.getConnectedUsers();
+ }
+
+ @ManagedOperation
+ public void sendMessage(String message) {
+ Message m = new Message(Message.OP.NOTICE);
+ m.data.put("notice", message);
+ connectionManager.broadcast(m);
+ }
+}
diff --git a/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/NotebookSocket.java b/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/NotebookSocket.java
new file mode 100644
index 0000000..dfaf858
--- /dev/null
+++ b/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/NotebookSocket.java
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.submarine.websocket;
+
+import org.apache.commons.lang.StringUtils;
+import org.eclipse.jetty.websocket.api.Session;
+import org.eclipse.jetty.websocket.api.WebSocketAdapter;
+
+import java.io.IOException;
+
+import javax.servlet.http.HttpServletRequest;
+
+/**
+ * Notebook websocket.
+ */
+public class NotebookSocket extends WebSocketAdapter {
+ private Session connection;
+ private NotebookSocketListener listener;
+ private HttpServletRequest request;
+ private String protocol;
+ private String user;
+
+ public NotebookSocket(HttpServletRequest req, String protocol,
+ NotebookSocketListener listener) {
+ this.listener = listener;
+ this.request = req;
+ this.protocol = protocol;
+ this.user = StringUtils.EMPTY;
+ }
+
+ @Override
+ public void onWebSocketClose(int closeCode, String message) {
+ listener.onClose(this, closeCode, message);
+ }
+
+ @Override
+ public void onWebSocketConnect(Session connection) {
+ this.connection = connection;
+ listener.onOpen(this);
+ }
+
+ @Override
+ public void onWebSocketText(String message) {
+ listener.onMessage(this, message);
+ }
+
+ public HttpServletRequest getRequest() {
+ return request;
+ }
+
+ public String getProtocol() {
+ return protocol;
+ }
+
+ public synchronized void send(String serializeMessage) throws IOException {
+ connection.getRemote().sendString(serializeMessage);
+ }
+
+ public String getUser() {
+ return user;
+ }
+
+ public void setUser(String user) {
+ this.user = user;
+ }
+
+ @Override
+ public String toString() {
+ return request.getRemoteHost() + ":" + request.getRemotePort();
+ }
+}
diff --git a/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/NotebookSocketListener.java b/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/NotebookSocketListener.java
new file mode 100644
index 0000000..0b9f318
--- /dev/null
+++ b/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/NotebookSocketListener.java
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.submarine.websocket;
+
+/**
+ * NoteboookSocket listener.
+ */
+public interface NotebookSocketListener {
+ void onClose(NotebookSocket socket, int code, String message);
+ void onOpen(NotebookSocket socket);
+ void onMessage(NotebookSocket socket, String message);
+}
diff --git a/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/NotebookWebSocketCreator.java b/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/NotebookWebSocketCreator.java
new file mode 100644
index 0000000..c834938
--- /dev/null
+++ b/submarine-workbench/workbench-server/src/main/java/org/apache/submarine/websocket/NotebookWebSocketCreator.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.submarine.websocket;
+
+import org.eclipse.jetty.websocket.servlet.ServletUpgradeRequest;
+import org.eclipse.jetty.websocket.servlet.ServletUpgradeResponse;
+import org.eclipse.jetty.websocket.servlet.WebSocketCreator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Responsible to create the WebSockets for the NotebookServer.
+ */
+public class NotebookWebSocketCreator implements WebSocketCreator {
+ private static final Logger LOG = LoggerFactory.getLogger(NotebookWebSocketCreator.class);
+ private NotebookServer notebookServer;
+
+ public NotebookWebSocketCreator(NotebookServer notebookServer) {
+ this.notebookServer = notebookServer;
+ }
+ public Object createWebSocket(ServletUpgradeRequest request, ServletUpgradeResponse response) {
+ return new NotebookSocket(request.getHttpServletRequest(), "", notebookServer);
+ }
+
+}
diff --git a/submarine-workbench/workbench-server/src/test/java/org/apache/submarine/server/AbstractWorkbenchServerTest.java b/submarine-workbench/workbench-server/src/test/java/org/apache/submarine/server/AbstractWorkbenchServerTest.java
new file mode 100644
index 0000000..681d20a
--- /dev/null
+++ b/submarine-workbench/workbench-server/src/test/java/org/apache/submarine/server/AbstractWorkbenchServerTest.java
@@ -0,0 +1,300 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.submarine.server;
+
+import com.google.gson.JsonElement;
+import com.google.gson.JsonParseException;
+import com.google.gson.JsonParser;
+import org.apache.commons.httpclient.HttpClient;
+import org.apache.commons.httpclient.methods.GetMethod;
+import org.apache.commons.io.FileUtils;
+import org.apache.submarine.commons.utils.SubmarineConfiguration;
+import org.apache.submarine.utils.TestUtils;
+import org.hamcrest.Description;
+import org.hamcrest.TypeSafeMatcher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+public abstract class AbstractWorkbenchServerTest {
+ protected static final Logger LOG =
+ LoggerFactory.getLogger(AbstractWorkbenchServerTest.class);
+
+ static final String WEBSOCKET_API_URL = "/ws";
+ static final String URL = getUrlToTest();
+ protected static final boolean WAS_RUNNING = checkIfServerIsRunning();
+
+ protected static File workbenchServerHome;
+ protected static File confDir;
+ protected static File notebookDir;
+
+ private String getUrl(String path) {
+ String url;
+ if (System.getProperty("url") != null) {
+ url = System.getProperty("url");
+ } else {
+ url = "http://localhost:8080";
+ }
+ if (path != null) {
+ url += path;
+ }
+ return url;
+ }
+
+ public static String getWebsocketApiUrlToTest() {
+ String websocketUrl = "ws://localhost:8080" + WEBSOCKET_API_URL;
+ if (System.getProperty("websocketUrl") != null) {
+ websocketUrl = System.getProperty("websocketurl");
+ }
+ return websocketUrl;
+ }
+
+ public static String getUrlToTest() {
+ String url = "http://localhost:8080";
+ if (System.getProperty("url") != null) {
+ url = System.getProperty("url");
+ }
+ return url;
+ }
+
+ static ExecutorService executor;
+ protected static final Runnable SERVER = new Runnable() {
+ @Override
+ public void run() {
+ try {
+ TestUtils.clearInstances();
+ WorkbenchServer.main(new String[]{""});
+ } catch (Exception e) {
+ LOG.error("Exception in WebDriverManager while getWebDriver ", e);
+ throw new RuntimeException(e);
+ }
+ }
+ };
+
+ private static void start(boolean withAuth, String testClassName,
+ boolean withKnox, boolean cleanData) throws Exception {
+ LOG.info("Starting WorkbenchServer withAuth: {}, testClassName: {}, withKnox: {}",
+ withAuth, testClassName, withKnox);
+
+ if (!WAS_RUNNING) {
+ // copy the resources files to a temp folder
+ workbenchServerHome = new File("..");
+ LOG.info("SUBMARINE_WORKBENCH_SERVER_HOME: "
+ + workbenchServerHome.getAbsolutePath());
+ confDir = new File(workbenchServerHome, "conf_" + testClassName);
+ confDir.mkdirs();
+
+ System.setProperty(SubmarineConfiguration.ConfVars.WORKBENCH_WEB_WAR.getVarName(),
+ new File("../workbench-web/dist").getAbsolutePath());
+ System.setProperty(SubmarineConfiguration.ConfVars.SUBMARINE_CONF_DIR.getVarName(),
+ confDir.getAbsolutePath());
+ notebookDir = new File(workbenchServerHome.getAbsolutePath() + "/notebook_" + testClassName);
+ if (cleanData) {
+ FileUtils.deleteDirectory(notebookDir);
+ }
+
+ // some test profile does not build workbench-web.
+ // to prevent submarine workbench server starting up fail,
+ // create workbench-web/dist directory
+ new File("../workbench-web/dist").mkdirs();
+
+ LOG.info("Staring test workbench server up...");
+
+ executor = Executors.newSingleThreadExecutor();
+ executor.submit(SERVER);
+ long s = System.currentTimeMillis();
+ boolean started = false;
+ while (System.currentTimeMillis() - s < 1000 * 60 * 3) { // 3 minutes
+ Thread.sleep(2000);
+ started = checkIfServerIsRunning();
+ if (started == true) {
+ break;
+ }
+ }
+ if (started == false) {
+ throw new RuntimeException("Can not start workbench server.");
+ }
+ LOG.info("Test workbench server stared.");
+ }
+ }
+
+ public static void startUpWithKnoxEnable(String testClassName) throws Exception {
+ start(true, testClassName, true, true);
+ }
+
+ public static void startUpWithAuthenticationEnable(String testClassName) throws Exception {
+ start(true, testClassName, false, true);
+ }
+
+ public static void startUp(String testClassName) throws Exception {
+ start(false, testClassName, false, true);
+ }
+
+ public static void startUp(String testClassName, boolean cleanData) throws Exception {
+ start(false, testClassName, false, cleanData);
+ }
+
+ private static String getHostname() {
+ try {
+ return InetAddress.getLocalHost().getHostName();
+ } catch (UnknownHostException e) {
+ LOG.error("Exception in WebDriverManager while getWebDriver ", e);
+ return "localhost";
+ }
+ }
+
+ public static void shutDown() throws Exception {
+ shutDown(true);
+ }
+
+ protected static void shutDown(final boolean deleteConfDir) throws Exception {
+ if (!WAS_RUNNING) {
+ LOG.info("Terminating test workbench server...");
+ WorkbenchServer.jettyWebServer.stop();
+ executor.shutdown();
+
+ long s = System.currentTimeMillis();
+ boolean started = true;
+ // 3 minutes
+ while (System.currentTimeMillis() - s < 1000 * 60 * 3) {
+ Thread.sleep(2000);
+ started = checkIfServerIsRunning();
+ if (started == false) {
+ break;
+ }
+ }
+ if (started == true) {
+ throw new RuntimeException("Can not stop Submarine workbench server");
+ }
+
+ LOG.info("Test Submarine workbench server terminated.");
+
+ if (deleteConfDir) {
+ FileUtils.deleteDirectory(confDir);
+ }
+ }
+ }
+
+ protected static GetMethod httpGet(String path) throws IOException {
+ return httpGet(path, "", "");
+ }
+
+ protected static GetMethod httpGet(String path, String user, String pwd) throws IOException {
+ return httpGet(path, user, pwd, "");
+ }
+
+ protected static GetMethod httpGet(String path, String user, String pwd, String cookies)
+ throws IOException {
+ LOG.info("Connecting to {}", URL + path);
+ HttpClient httpClient = new HttpClient();
+ GetMethod getMethod = new GetMethod(URL + path);
+ getMethod.addRequestHeader("Origin", URL);
+ httpClient.executeMethod(getMethod);
+ LOG.info("{} - {}", getMethod.getStatusCode(), getMethod.getStatusText());
+ return getMethod;
+ }
+
+ protected static boolean checkIfServerIsRunning() {
+ GetMethod request = null;
+ boolean isRunning = false;
+ try {
+ request = httpGet("/");
+ isRunning = request.getStatusCode() == 200;
+ } catch (IOException e) {
+ LOG.warn("AbstractTestRestApi.checkIfServerIsRunning() fails .. " +
+ "Submarine workbench server is not running");
+ isRunning = false;
+ } finally {
+ if (request != null) {
+ request.releaseConnection();
+ }
+ }
+ return isRunning;
+ }
+
+ protected TypeSafeMatcher<String> isJSON() {
+ return new TypeSafeMatcher<String>() {
+ @Override
+ public boolean matchesSafely(String body) {
+ String b = body.trim();
+ return (b.startsWith("{") && b.endsWith("}")) || (b.startsWith("[") && b.endsWith("]"));
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description.appendText("response in JSON format ");
+ }
+
+ @Override
+ protected void describeMismatchSafely(String item, Description description) {
+ description.appendText("got ").appendText(item);
+ }
+ };
+ }
+
+ protected TypeSafeMatcher<String> isValidJSON() {
+ return new TypeSafeMatcher<String>() {
+ @Override
+ public boolean matchesSafely(String body) {
+ boolean isValid = true;
+ try {
+ new JsonParser().parse(body);
+ } catch (JsonParseException e) {
+ LOG.error("Exception in AbstractTestRestApi while matchesSafely ", e);
+ isValid = false;
+ }
+ return isValid;
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description.appendText("response in JSON format ");
+ }
+
+ @Override
+ protected void describeMismatchSafely(String item, Description description) {
+ description.appendText("got ").appendText(item);
+ }
+ };
+ }
+
+ protected TypeSafeMatcher<? super JsonElement> hasRootElementNamed(final String memberName) {
+ return new TypeSafeMatcher<JsonElement>() {
+ @Override
+ protected boolean matchesSafely(JsonElement item) {
+ return item.isJsonObject() && item.getAsJsonObject().has(memberName);
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description.appendText("response in JSON format with \"").appendText(memberName)
+ .appendText("\" beeing a root element ");
+ }
+
+ @Override
+ protected void describeMismatchSafely(JsonElement root, Description description) {
+ description.appendText("got ").appendText(root.toString());
+ }
+ };
+ }
+}
diff --git a/submarine-workbench/workbench-server/src/test/java/org/apache/submarine/server/JsonResponseTest.java b/submarine-workbench/workbench-server/src/test/java/org/apache/submarine/server/JsonResponseTest.java
index 67f261f..ec7b78f 100644
--- a/submarine-workbench/workbench-server/src/test/java/org/apache/submarine/server/JsonResponseTest.java
+++ b/submarine-workbench/workbench-server/src/test/java/org/apache/submarine/server/JsonResponseTest.java
@@ -1,21 +1,20 @@
/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ * 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.
+ * 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 org.apache.submarine.server;
import com.google.gson.Gson;
diff --git a/submarine-workbench/workbench-server/src/test/java/org/apache/submarine/websocket/NotebookServerTest.java b/submarine-workbench/workbench-server/src/test/java/org/apache/submarine/websocket/NotebookServerTest.java
new file mode 100644
index 0000000..7fd706b
--- /dev/null
+++ b/submarine-workbench/workbench-server/src/test/java/org/apache/submarine/websocket/NotebookServerTest.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.submarine.websocket;
+
+import org.apache.submarine.server.AbstractWorkbenchServerTest;
+import org.eclipse.jetty.websocket.api.Session;
+import org.eclipse.jetty.websocket.api.StatusCode;
+import org.eclipse.jetty.websocket.api.WebSocketAdapter;
+import org.eclipse.jetty.websocket.client.WebSocketClient;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.net.URI;
+import java.util.concurrent.Future;
+
+public class NotebookServerTest {
+
+ @BeforeClass
+ public static void init() throws Exception {
+ AbstractWorkbenchServerTest.startUp(
+ NotebookServerTest.class.getSimpleName());
+ }
+
+ @AfterClass
+ public static void destroy() throws Exception {
+ AbstractWorkbenchServerTest.shutDown();
+ }
+
+ @Test
+ public void testWebsocketConnection() throws Exception{
+ URI uri = URI.create(
+ AbstractWorkbenchServerTest.getWebsocketApiUrlToTest());
+ WebSocketClient client = new WebSocketClient();
+ try {
+ client.start();
+ // The socket that receives events
+ EventSocket socket = new EventSocket();
+ // Attempt Connect
+ Future<Session> fut = client.connect(socket, uri);
+ // Wait for Connect
+ Session session = fut.get();
+ // Send a message
+ session.getRemote().sendString("Hello");
+ // Close session
+ //session.close();
+ session.close(StatusCode.NORMAL, "I'm done");
+ } finally {
+ client.stop();
+ }
+ }
+
+ public class EventSocket extends WebSocketAdapter
+ {
+ @Override
+ public void onWebSocketConnect(Session sess)
+ {
+ super.onWebSocketConnect(sess);
+ System.out.println("Socket Connected: " + sess);
+ }
+
+ @Override
+ public void onWebSocketText(String message)
+ {
+ super.onWebSocketText(message);
+ System.out.println("Received TEXT message: " + message);
+ }
+
+ @Override
+ public void onWebSocketClose(int statusCode, String reason)
+ {
+ super.onWebSocketClose(statusCode, reason);
+ System.out.println("Socket Closed: [" + statusCode + "] " + reason);
+ }
+
+ @Override
+ public void onWebSocketError(Throwable cause)
+ {
+ super.onWebSocketError(cause);
+ cause.printStackTrace(System.err);
+ }
+ }
+}