Add support for websocket subscriptions

src/main/java/ca/uhn/fhir/jpa/starter/ApplicationContext.java

@@ -0,0 +1,26 @@
+package ca.uhn.fhir.jpa.starter;
+
+import ca.uhn.fhir.context.FhirVersionEnum;
+import org.springframework.web.context.support.AnnotationConfigWebApplicationContext;
+
+public class ApplicationContext extends AnnotationConfigWebApplicationContext {
+
+    public ApplicationContext() {
+        FhirVersionEnum fhirVersion = HapiProperties.getFhirVersion();
+        if (fhirVersion == FhirVersionEnum.DSTU2) {
+            register(FhirServerConfigDstu2.class, FhirServerConfigCommon.class);
+        } else if (fhirVersion == FhirVersionEnum.DSTU3) {
+            register(FhirServerConfigDstu3.class, FhirServerConfigCommon.class);
+        } else if (fhirVersion == FhirVersionEnum.R4) {
+            register(FhirServerConfigR4.class, FhirServerConfigCommon.class);
+        } else {
+            throw new IllegalStateException();
+        }
+
+        if (HapiProperties.getSubscriptionWebsocketEnabled()) {
+            register(ca.uhn.fhir.jpa.config.WebsocketDispatcherConfig.class);
+        }
+
+    }
+
+}

src/main/java/ca/uhn/fhir/jpa/starter/FhirServerConfigCommon.java

@@ -53,18 +53,21 @@ public class FhirServerConfigCommon {
 		retVal.setFetchSizeDefaultMaximum(maxFetchSize);
 		ourLog.info("Server configured to have a maximum fetch size of " + (maxFetchSize == Integer.MAX_VALUE? "'unlimited'": maxFetchSize));
 		
-		// You can enable these if you want to support Subscriptions from your server
+		// Subscriptions are enabled by channel type
 		if (HapiProperties.getSubscriptionRestHookEnabled()) {
 			ourLog.info("Enabling REST-hook subscriptions");
 			retVal.addSupportedSubscriptionType(Subscription.SubscriptionChannelType.RESTHOOK);
 		}
-
 		if (HapiProperties.getSubscriptionEmailEnabled()) {
 			ourLog.info("Enabling email subscriptions");
 			retVal.addSupportedSubscriptionType(Subscription.SubscriptionChannelType.EMAIL);
 		}
+		if (HapiProperties.getSubscriptionWebsocketEnabled()) {
+			ourLog.info("Enabling websocket subscriptions");
+			retVal.addSupportedSubscriptionType(Subscription.SubscriptionChannelType.WEBSOCKET);
+		}
 
-    	return retVal;
+		return retVal;
 	}
 
 	@Bean

src/main/java/ca/uhn/fhir/jpa/starter/FhirServerConfigR4.java

@@ -18,6 +18,7 @@ public class FhirServerConfigR4 extends BaseJavaConfigR4 {
     @Autowired
     private DataSource myDataSource;
 
+
     /**
      * We override the paging provider definition so that we can customize
      * the default/max page sizes for search results. You can set these however

src/main/java/ca/uhn/fhir/jpa/starter/HapiProperties.java

@@ -38,6 +38,7 @@ public class HapiProperties {
     static final String SERVER_NAME = "server.name";
     static final String SUBSCRIPTION_EMAIL_ENABLED = "subscription.email.enabled";
     static final String SUBSCRIPTION_RESTHOOK_ENABLED = "subscription.resthook.enabled";
+    static final String SUBSCRIPTION_WEBSOCKET_ENABLED = "subscription.websocket.enabled";
     static final String TEST_PORT = "test.port";
 
     private static Properties properties;
@@ -234,11 +235,11 @@ public class HapiProperties {
     }
 
     public static Boolean getAllowMultipleDelete() {
-        return HapiProperties.getBooleanProperty(ALLOW_MULTIPLE_DELETE, true);
+        return HapiProperties.getBooleanProperty(ALLOW_MULTIPLE_DELETE, false);
     }
 
     public static Boolean getAllowExternalReferences() {
-        return HapiProperties.getBooleanProperty(ALLOW_EXTERNAL_REFERENCES, true);
+        return HapiProperties.getBooleanProperty(ALLOW_EXTERNAL_REFERENCES, false);
     }
 
     public static Boolean getExpungeEnabled() {
@@ -266,10 +267,14 @@ public class HapiProperties {
     }
 
     public static Boolean getSubscriptionEmailEnabled() {
-        return HapiProperties.getBooleanProperty(SUBSCRIPTION_EMAIL_ENABLED, true);
+        return HapiProperties.getBooleanProperty(SUBSCRIPTION_EMAIL_ENABLED, false);
     }
 
     public static Boolean getSubscriptionRestHookEnabled() {
-        return HapiProperties.getBooleanProperty(SUBSCRIPTION_RESTHOOK_ENABLED, true);
+        return HapiProperties.getBooleanProperty(SUBSCRIPTION_RESTHOOK_ENABLED, false);
+    }
+
+    public static Boolean getSubscriptionWebsocketEnabled() {
+        return HapiProperties.getBooleanProperty(SUBSCRIPTION_WEBSOCKET_ENABLED, false);
     }
 }

src/main/java/ca/uhn/fhir/jpa/starter/JpaRestfulServer.java

@@ -4,6 +4,7 @@ import ca.uhn.fhir.context.FhirContext;
 import ca.uhn.fhir.context.FhirVersionEnum;
 import ca.uhn.fhir.jpa.dao.DaoConfig;
 import ca.uhn.fhir.jpa.dao.IFhirSystemDao;
+import ca.uhn.fhir.jpa.model.interceptor.api.IInterceptorRegistry;
 import ca.uhn.fhir.jpa.provider.JpaConformanceProviderDstu2;
 import ca.uhn.fhir.jpa.provider.JpaSystemProviderDstu2;
 import ca.uhn.fhir.jpa.provider.SubscriptionTriggeringProvider;
@@ -14,6 +15,8 @@ import ca.uhn.fhir.jpa.provider.r4.JpaConformanceProviderR4;
 import ca.uhn.fhir.jpa.provider.r4.JpaSystemProviderR4;
 import ca.uhn.fhir.jpa.provider.r4.TerminologyUploaderProviderR4;
 import ca.uhn.fhir.jpa.search.DatabaseBackedPagingProvider;
+import ca.uhn.fhir.jpa.subscription.SubscriptionActivatingInterceptor;
+import ca.uhn.fhir.jpa.subscription.SubscriptionMatcherInterceptor;
 import ca.uhn.fhir.model.dstu2.composite.MetaDt;
 import ca.uhn.fhir.narrative.DefaultThymeleafNarrativeGenerator;
 import ca.uhn.fhir.rest.server.HardcodedServerAddressStrategy;
@@ -22,7 +25,7 @@ import ca.uhn.fhir.rest.server.RestfulServer;
 import ca.uhn.fhir.rest.server.interceptor.ResponseHighlighterInterceptor;
 import org.hl7.fhir.dstu3.model.Bundle;
 import org.hl7.fhir.dstu3.model.Meta;
-import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.ApplicationContext;
 
 import javax.servlet.ServletException;
 import java.util.List;
@@ -30,12 +33,6 @@ import java.util.List;
 public class JpaRestfulServer extends RestfulServer {
 
     private static final long serialVersionUID = 1L;
-    private AnnotationConfigApplicationContext appCtx;
-
-    @Override
-    public void destroy() {
-        appCtx.close();
-    }
 
     @SuppressWarnings("unchecked")
     @Override
@@ -46,27 +43,25 @@ public class JpaRestfulServer extends RestfulServer {
          * Create a FhirContext object that uses the version of FHIR
          * specified in the properties file.
          */
-        FhirVersionEnum fhirVersion = HapiProperties.getFhirVersion();
+        ApplicationContext appCtx = (ApplicationContext) getServletContext().getAttribute("org.springframework.web.context.WebApplicationContext.ROOT");
-        appCtx = new AnnotationConfigApplicationContext();
+
+//        if (HapiProperties.getSubscriptionWebsocketEnabled()) {
+//            appCtx.register(WebsocketDispatcherConfig.class);
+//        }
 
         /*
          * ResourceProviders are fetched from the Spring context
          */
+        FhirVersionEnum fhirVersion = HapiProperties.getFhirVersion();
         List<IResourceProvider> resourceProviders;
         Object systemProvider;
         if (fhirVersion == FhirVersionEnum.DSTU2) {
-            appCtx.register(FhirServerConfigDstu2.class, FhirServerConfigCommon.class);
-            appCtx.refresh();
             resourceProviders = appCtx.getBean("myResourceProvidersDstu2", List.class);
             systemProvider = appCtx.getBean("mySystemProviderDstu2", JpaSystemProviderDstu2.class);
         } else if (fhirVersion == FhirVersionEnum.DSTU3) {
-            appCtx.register(FhirServerConfigDstu3.class, FhirServerConfigCommon.class);
-            appCtx.refresh();
             resourceProviders = appCtx.getBean("myResourceProvidersDstu3", List.class);
             systemProvider = appCtx.getBean("mySystemProviderDstu3", JpaSystemProviderDstu3.class);
         } else if (fhirVersion == FhirVersionEnum.R4) {
-            appCtx.register(FhirServerConfigR4.class, FhirServerConfigCommon.class);
-            appCtx.refresh();
             resourceProviders = appCtx.getBean("myResourceProvidersR4", List.class);
             systemProvider = appCtx.getBean("mySystemProviderR4", JpaSystemProviderR4.class);
         } else {
@@ -174,6 +169,23 @@ public class JpaRestfulServer extends RestfulServer {
             SubscriptionTriggeringProvider retriggeringProvider = appCtx.getBean(SubscriptionTriggeringProvider.class);
             registerProvider(retriggeringProvider);
         }
+
+        // If subscriptions are enabled, we want to register the interceptor that
+        // will activate them and match results against them
+        if (HapiProperties.getSubscriptionWebsocketEnabled() ||
+                HapiProperties.getSubscriptionEmailEnabled() ||
+                HapiProperties.getSubscriptionRestHookEnabled()) {
+            IInterceptorRegistry interceptorRegistry = appCtx.getBean(IInterceptorRegistry.class);
+
+            SubscriptionActivatingInterceptor subscriptionActivatingInterceptor = appCtx.getBean(SubscriptionActivatingInterceptor.class);
+            interceptorRegistry.registerInterceptor(subscriptionActivatingInterceptor);
+
+            SubscriptionMatcherInterceptor subscriptionMatcherInterceptor = appCtx.getBean(SubscriptionMatcherInterceptor.class);
+            subscriptionMatcherInterceptor.start();
+            interceptorRegistry.registerInterceptor(subscriptionMatcherInterceptor);
+        }
+
+
     }
 
 }

src/main/resources/hapi.properties

@@ -61,3 +61,6 @@ subscription.resthook.enabled=true
 
 # Enable Email Subscription Channel
 subscription.email.enabled=false
+
+# Enable Websocket Subscription Channel
+subscription.websocket.enabled=false

src/main/webapp/WEB-INF/web.xml

@@ -5,23 +5,20 @@
 		metadata-complete="false"
 		version="3.1">
 
-	<!--
 	<listener>
 		<listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
 	</listener>
 	<context-param>
 		<param-name>contextClass</param-name>
 		<param-value>
-			org.springframework.web.context.support.AnnotationConfigWebApplicationContext
+			ca.uhn.fhir.jpa.starter.ApplicationContext
 		</param-value>
 	</context-param>
 	<context-param>
 		<param-name>contextConfigLocation</param-name>
 		<param-value>
-			ca.uhn.fhir.jpa.starter.FhirServerConfig
 		</param-value>
 	</context-param>
-	-->
 
 	<!-- Servlets -->
 	<servlet>
@@ -33,7 +30,9 @@
 		</init-param>
 		<init-param>
 			<param-name>contextConfigLocation</param-name>
-			<param-value>ca.uhn.fhir.jpa.starter.FhirTesterConfig</param-value>
+			<param-value>
+				ca.uhn.fhir.jpa.starter.FhirTesterConfig,
+			</param-value>
 		</init-param>
 		<load-on-startup>2</load-on-startup>
 	</servlet>

src/test/java/ca/uhn/fhir/jpa/starter/ExampleServerDstu2IT.java

@@ -31,6 +31,7 @@ public class ExampleServerDstu2IT {
 	static {
 		HapiProperties.forceReload();
 		HapiProperties.setProperty(HapiProperties.FHIR_VERSION, "DSTU2");
+		HapiProperties.setProperty(HapiProperties.DATASOURCE_URL, "jdbc:derby:memory:dbr2;create=true");
 		HapiProperties.setProperty(HapiProperties.TEST_PORT, Integer.toString(PortUtil.findFreePort()));
 		ourCtx = FhirContext.forDstu2();
 		ourPort = HapiProperties.getTestPort();

src/test/java/ca/uhn/fhir/jpa/starter/ExampleServerDstu3IT.java

@@ -30,6 +30,7 @@ public class ExampleServerDstu3IT {
 	static {
 		HapiProperties.forceReload();
 		HapiProperties.setProperty(HapiProperties.FHIR_VERSION, "DSTU3");
+		HapiProperties.setProperty(HapiProperties.DATASOURCE_URL, "jdbc:derby:memory:dbr3;create=true");
 		HapiProperties.setProperty(HapiProperties.TEST_PORT, Integer.toString(PortUtil.findFreePort()));
 		ourCtx = FhirContext.forDstu3();
 		ourPort = HapiProperties.getTestPort();

src/test/java/ca/uhn/fhir/jpa/starter/ExampleServerR4IT.java

@@ -1,94 +1,170 @@
 package ca.uhn.fhir.jpa.starter;
 
 import ca.uhn.fhir.context.FhirContext;
+import ca.uhn.fhir.rest.api.CacheControlDirective;
+import ca.uhn.fhir.rest.api.EncodingEnum;
+import ca.uhn.fhir.rest.api.MethodOutcome;
 import ca.uhn.fhir.rest.client.api.IGenericClient;
 import ca.uhn.fhir.rest.client.api.ServerValidationModeEnum;
 import ca.uhn.fhir.rest.client.interceptor.LoggingInterceptor;
 import ca.uhn.fhir.util.PortUtil;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.webapp.WebAppContext;
-import org.hl7.fhir.r4.model.Patient;
+import org.eclipse.jetty.websocket.api.Session;
+import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
+import org.eclipse.jetty.websocket.client.WebSocketClient;
 import org.hl7.fhir.instance.model.api.IIdType;
+import org.hl7.fhir.r4.model.Bundle;
+import org.hl7.fhir.r4.model.Observation;
+import org.hl7.fhir.r4.model.Patient;
+import org.hl7.fhir.r4.model.Subscription;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
 import java.io.File;
 import java.io.IOException;
+import java.net.URI;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 
+import static ca.uhn.fhir.util.TestUtil.waitForSize;
 import static org.junit.Assert.assertEquals;
 
 public class ExampleServerR4IT {
 
-	private static final org.slf4j.Logger ourLog = org.slf4j.LoggerFactory.getLogger(ExampleServerR4IT.class);
+    private static final org.slf4j.Logger ourLog = org.slf4j.LoggerFactory.getLogger(ExampleServerR4IT.class);
-	private static IGenericClient ourClient;
+    private static IGenericClient ourClient;
-	private static FhirContext ourCtx;
+    private static FhirContext ourCtx;
-	private static int ourPort;
+    private static int ourPort;
 
-	private static Server ourServer;
+    private static Server ourServer;
-	private static String ourServerBase;
+    private static String ourServerBase;
 
-	static {
+    static {
-		HapiProperties.forceReload();
+        HapiProperties.forceReload();
-		HapiProperties.setProperty(HapiProperties.FHIR_VERSION, "R4");
+        HapiProperties.setProperty(HapiProperties.DATASOURCE_URL, "jdbc:derby:memory:dbr4;create=true");
-		HapiProperties.setProperty(HapiProperties.TEST_PORT, Integer.toString(PortUtil.findFreePort()));
+        HapiProperties.setProperty(HapiProperties.FHIR_VERSION, "R4");
-		ourCtx = FhirContext.forR4();
+        HapiProperties.setProperty(HapiProperties.TEST_PORT, Integer.toString(PortUtil.findFreePort()));
-		ourPort = HapiProperties.getTestPort();
+        HapiProperties.setProperty(HapiProperties.SUBSCRIPTION_WEBSOCKET_ENABLED, "true");
-	}
+        ourCtx = FhirContext.forR4();
+        ourPort = HapiProperties.getTestPort();
+    }
 
-	@Test
+    @Test
-	public void testCreateAndRead() throws IOException {
+    public void testCreateAndRead() throws IOException {
-		ourLog.info("Base URL is: http://localhost:" + ourPort + HapiProperties.getServerBase());
+        ourLog.info("Base URL is: http://localhost:" + ourPort + HapiProperties.getServerBase());
-		String methodName = "testCreateResourceConditional";
+        String methodName = "testCreateResourceConditional";
 
-		Patient pt = new Patient();
+        Patient pt = new Patient();
-		pt.addName().setFamily(methodName);
+        pt.addName().setFamily(methodName);
-		IIdType id = ourClient.create().resource(pt).execute().getId();
+        IIdType id = ourClient.create().resource(pt).execute().getId();
 
-		Patient pt2 = ourClient.read().resource(Patient.class).withId(id).execute();
+        Patient pt2 = ourClient.read().resource(Patient.class).withId(id).execute();
-		assertEquals(methodName, pt2.getName().get(0).getFamily());
+        assertEquals(methodName, pt2.getName().get(0).getFamily());
-	}
+    }
 
-	@AfterClass
-	public static void afterClass() throws Exception {
-		ourServer.stop();
-	}
 
-	@BeforeClass
+    @Test
-	public static void beforeClass() throws Exception {
+    public void testWebsocketSubscription() throws Exception {
-		/*
+        /*
-		 * This runs under maven, and I'm not sure how else to figure out the target directory from code..
+         * Create subscription
-		 */
+         */
-		String path = ExampleServerR4IT.class.getClassLoader().getResource(".keep_hapi-fhir-jpaserver-starter").getPath();
+        Subscription subscription = new Subscription();
-		path = new File(path).getParent();
+        subscription.setReason("Monitor new neonatal function (note, age will be determined by the monitor)");
-		path = new File(path).getParent();
+        subscription.setStatus(Subscription.SubscriptionStatus.REQUESTED);
-		path = new File(path).getParent();
+        subscription.setCriteria("Observation?status=final");
 
-		ourLog.info("Project base path is: {}", path);
+        Subscription.SubscriptionChannelComponent channel = new Subscription.SubscriptionChannelComponent();
+        channel.setType(Subscription.SubscriptionChannelType.WEBSOCKET);
+        channel.setPayload("application/json");
+        subscription.setChannel(channel);
 
-		if (ourPort == 0) {
+        MethodOutcome methodOutcome = ourClient.create().resource(subscription).execute();
-			ourPort = RandomServerPortProvider.findFreePort();
+        IIdType mySubscriptionId = methodOutcome.getId();
-		}
-		ourServer = new Server(ourPort);
 
-		WebAppContext webAppContext = new WebAppContext();
+        // Wait for the subscription to be activated
-		webAppContext.setContextPath("/");
+        waitForSize(1, () -> ourClient.search().forResource(Subscription.class).where(Subscription.STATUS.exactly().code("active")).cacheControl(new CacheControlDirective().setNoCache(true)).returnBundle(Bundle.class).execute().getEntry().size());
-		webAppContext.setDescriptor(path + "/src/main/webapp/WEB-INF/web.xml");
-		webAppContext.setResourceBase(path + "/target/hapi-fhir-jpaserver-starter");
-		webAppContext.setParentLoaderPriority(true);
 
-		ourServer.setHandler(webAppContext);
+        /*
-		ourServer.start();
+         * Attach websocket
+         */
 
-		ourCtx.getRestfulClientFactory().setServerValidationMode(ServerValidationModeEnum.NEVER);
+        WebSocketClient myWebSocketClient = new WebSocketClient();
-		ourCtx.getRestfulClientFactory().setSocketTimeout(1200 * 1000);
+        SocketImplementation mySocketImplementation = new SocketImplementation(mySubscriptionId.getIdPart(), EncodingEnum.JSON);
-		ourServerBase = "http://localhost:" + ourPort + HapiProperties.getServerBase();
-		ourClient = ourCtx.newRestfulGenericClient(ourServerBase);
-		ourClient.registerInterceptor(new LoggingInterceptor(true));
-	}
 
-	public static void main(String[] theArgs) throws Exception {
+        myWebSocketClient.start();
-		ourPort = 8080;
+        URI echoUri = new URI("ws://localhost:" + ourPort + "/websocket");
-		beforeClass();
+        ClientUpgradeRequest request = new ClientUpgradeRequest();
-	}
+        ourLog.info("Connecting to : {}", echoUri);
+        Future<Session> connection = myWebSocketClient.connect(mySocketImplementation, echoUri, request);
+        Session session = connection.get(2, TimeUnit.SECONDS);
+
+        ourLog.info("Connected to WS: {}", session.isOpen());
+
+        /*
+         * Create a matching resource
+         */
+        Observation obs = new Observation();
+        obs.setStatus(Observation.ObservationStatus.FINAL);
+        ourClient.create().resource(obs).execute();
+
+        // Give some time for the subscription to deliver
+        Thread.sleep(2000);
+
+        /*
+         * Ensure that we receive a ping on the websocket
+         */
+        waitForSize(1, () -> mySocketImplementation.myPingCount);
+
+        /*
+         * Clean up
+         */
+        ourClient.delete().resourceById(mySubscriptionId).execute();
+    }
+
+
+    @AfterClass
+    public static void afterClass() throws Exception {
+        ourServer.stop();
+    }
+
+    @BeforeClass
+    public static void beforeClass() throws Exception {
+        /*
+         * This runs under maven, and I'm not sure how else to figure out the target directory from code..
+         */
+        String path = ExampleServerR4IT.class.getClassLoader().getResource(".keep_hapi-fhir-jpaserver-starter").getPath();
+        path = new File(path).getParent();
+        path = new File(path).getParent();
+        path = new File(path).getParent();
+
+        ourLog.info("Project base path is: {}", path);
+
+        if (ourPort == 0) {
+            ourPort = RandomServerPortProvider.findFreePort();
+        }
+        ourServer = new Server(ourPort);
+
+        WebAppContext webAppContext = new WebAppContext();
+        webAppContext.setContextPath("/");
+        webAppContext.setDisplayName("HAPI FHIR");
+        webAppContext.setDescriptor(path + "/src/main/webapp/WEB-INF/web.xml");
+        webAppContext.setResourceBase(path + "/target/hapi-fhir-jpaserver-starter");
+        webAppContext.setParentLoaderPriority(true);
+
+        ourServer.setHandler(webAppContext);
+        ourServer.start();
+
+        ourCtx.getRestfulClientFactory().setServerValidationMode(ServerValidationModeEnum.NEVER);
+        ourCtx.getRestfulClientFactory().setSocketTimeout(1200 * 1000);
+        ourServerBase = "http://localhost:" + ourPort + HapiProperties.getServerBase();
+        ourClient = ourCtx.newRestfulGenericClient(ourServerBase);
+        ourClient.registerInterceptor(new LoggingInterceptor(true));
+    }
+
+    public static void main(String[] theArgs) throws Exception {
+        ourPort = 8080;
+        beforeClass();
+    }
 }

src/test/java/ca/uhn/fhir/jpa/starter/SocketImplementation.java

@@ -0,0 +1,88 @@
+
+package ca.uhn.fhir.jpa.starter;
+
+import ca.uhn.fhir.rest.api.EncodingEnum;
+import org.eclipse.jetty.websocket.api.Session;
+import org.eclipse.jetty.websocket.api.annotations.OnWebSocketConnect;
+import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage;
+import org.eclipse.jetty.websocket.api.annotations.WebSocket;
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@WebSocket
+public class SocketImplementation {
+
+	private static final Logger ourLog = org.slf4j.LoggerFactory.getLogger(SocketImplementation.class);
+	private String myCriteria;
+	protected String myError;
+	protected boolean myGotBound;
+	private List<String> myMessages = new ArrayList<String>();
+	protected int myPingCount;
+	protected String mySubsId;
+	private Session session;
+
+	public SocketImplementation(String theCriteria, EncodingEnum theEncoding) {
+		myCriteria = theCriteria;
+	}
+
+	public List<String> getMessages() {
+		return myMessages;
+	}
+
+	public void keepAlive() {
+		if (this.session != null) {
+			try {
+				session.getRemote().sendString("keep alive");
+			} catch (Throwable t) {
+				ourLog.error("Failure", t);
+			}
+		}
+	}
+
+	/**
+	 * This method is executed when the client is connecting to the server.
+	 * In this case, we are sending a message to create the subscription dynamiclly
+	 *
+	 * @param session
+	 */
+	@OnWebSocketConnect
+	public void onConnect(Session session) {
+		ourLog.info("Got connect: {}", session);
+		this.session = session;
+		try {
+			String sending = "bind " + myCriteria;
+			ourLog.info("Sending: {}", sending);
+			session.getRemote().sendString(sending);
+
+			ourLog.info("Connection: DONE");
+		} catch (Throwable t) {
+			t.printStackTrace();
+			ourLog.error("Failure", t);
+		}
+	}
+
+	/**
+	 * This is the message handler for the client
+	 *
+	 * @param theMsg
+	 */
+	@OnWebSocketMessage
+	public void onMessage(String theMsg) {
+		ourLog.info("Got msg: " + theMsg);
+		myMessages.add(theMsg);
+
+		if (theMsg.startsWith("bound ")) {
+			myGotBound = true;
+			mySubsId = (theMsg.substring("bound ".length()));
+		} else if (myGotBound && theMsg.startsWith("add " + mySubsId + "\n")) {
+			String text = theMsg.substring(("add " + mySubsId + "\n").length());
+			ourLog.info("text: " + text);
+		} else if (theMsg.startsWith("ping ")) {
+			myPingCount++;
+		} else {
+			myError = "Unexpected message: " + theMsg;
+		}
+	}
+}