Merge remote-tracking branch 'upstream/master'
# Conflicts: # src/main/java/ca/uhn/fhir/jpa/starter/JpaRestfulServer.java # src/main/resources/hapi.properties # src/test/java/ca/uhn/fhir/jpa/starter/ExampleServerR4IT.java
This commit is contained in:
@@ -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);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -53,16 +53,19 @@ 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;
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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";
|
||||
static final String TESTER_CONFIG_REFUSE_TO_FETCH_THIRD_PARTY_URLS = "tester.config.refuse_to_fetch_third_party_urls";
|
||||
|
||||
@@ -235,11 +236,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() {
|
||||
@@ -271,10 +272,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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ 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.executor.InterceptorService;
|
||||
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;
|
||||
@@ -17,6 +18,8 @@ import ca.uhn.fhir.jpa.provider.r4.TerminologyUploaderProviderR4;
|
||||
import ca.uhn.fhir.jpa.search.DatabaseBackedPagingProvider;
|
||||
import ca.uhn.fhir.jpa.subscription.SubscriptionInterceptorLoader;
|
||||
import ca.uhn.fhir.jpa.subscription.module.interceptor.SubscriptionDebugLogInterceptor;
|
||||
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;
|
||||
@@ -26,8 +29,8 @@ import ca.uhn.fhir.rest.server.interceptor.CorsInterceptor;
|
||||
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.web.cors.CorsConfiguration;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
|
||||
import javax.servlet.ServletException;
|
||||
import java.util.Arrays;
|
||||
@@ -36,12 +39,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
|
||||
@@ -52,34 +49,29 @@ public class JpaRestfulServer extends RestfulServer {
|
||||
* Create a FhirContext object that uses the version of FHIR
|
||||
* specified in the properties file.
|
||||
*/
|
||||
FhirVersionEnum fhirVersion = HapiProperties.getFhirVersion();
|
||||
setFhirContext(new FhirContext(fhirVersion));
|
||||
|
||||
appCtx = new AnnotationConfigApplicationContext();
|
||||
ApplicationContext appCtx = (ApplicationContext) getServletContext().getAttribute("org.springframework.web.context.WebApplicationContext.ROOT");
|
||||
|
||||
/*
|
||||
* 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 {
|
||||
throw new IllegalStateException();
|
||||
}
|
||||
|
||||
setFhirContext(appCtx.getBean(FhirContext.class));
|
||||
|
||||
registerProviders(resourceProviders);
|
||||
registerProvider(systemProvider);
|
||||
|
||||
@@ -200,12 +192,23 @@ public class JpaRestfulServer extends RestfulServer {
|
||||
CorsInterceptor interceptor = new CorsInterceptor(config);
|
||||
registerInterceptor(interceptor);
|
||||
|
||||
// Enable the use of subscriptions
|
||||
// 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()) {
|
||||
|
||||
// Loads subscription interceptors (SubscriptionActivatingInterceptor, SubscriptionMatcherInterceptor)
|
||||
// with activation of scheduled subscription
|
||||
SubscriptionInterceptorLoader subscriptionInterceptorLoader = appCtx.getBean(SubscriptionInterceptorLoader.class);
|
||||
subscriptionInterceptorLoader.registerInterceptors();
|
||||
|
||||
// Subscription debug logging
|
||||
InterceptorService interceptorService = (InterceptorService) appCtx.getBean("interceptorService");
|
||||
interceptorService.registerInterceptor(new SubscriptionDebugLogInterceptor());
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -8,10 +8,16 @@ fhir_version=R4
|
||||
# server, put the DNS name of that server here.
|
||||
server_address=http://localhost:8080/fhir/
|
||||
|
||||
# For Jetty, use this:
|
||||
# server_address=http://localhost:8080/hapi-fhir-jpaserver/fhir/
|
||||
|
||||
# This is the context path for the FHIR endpoint. If this is changed, the
|
||||
# setting above should also be changed.
|
||||
server.base=/fhir
|
||||
|
||||
# For Jetty, use this:
|
||||
# server.base=/hapi-fhir-jpaserver/fhir
|
||||
|
||||
default_encoding=JSON
|
||||
etag_support=ENABLED
|
||||
default_page_size=20
|
||||
@@ -32,8 +38,6 @@ datasource.password=fhirPass
|
||||
server.name=Local Tester
|
||||
server.id=home
|
||||
test.port=
|
||||
subscription.email.enabled=true
|
||||
subscription.resthook.enabled=true
|
||||
hibernate.dialect=org.hibernate.dialect.MySQL5Dialect
|
||||
hibernate.search.model_mapping=ca.uhn.fhir.jpa.search.LuceneSearchMappingFactory
|
||||
hibernate.format_sql=false
|
||||
@@ -48,3 +52,17 @@ hibernate.search.default.directory_provider=filesystem
|
||||
hibernate.search.default.indexBase=target/lucenefiles
|
||||
hibernate.search.lucene_version=LUCENE_CURRENT
|
||||
tester.config.refuse_to_fetch_third_party_urls=false
|
||||
|
||||
|
||||
##################################################
|
||||
# Subscriptions
|
||||
##################################################
|
||||
|
||||
# Enable REST Hook Subscription Channel
|
||||
subscription.resthook.enabled=true
|
||||
|
||||
# Enable Email Subscription Channel
|
||||
subscription.email.enabled=false
|
||||
|
||||
# Enable Websocket Subscription Channel
|
||||
subscription.websocket.enabled=false
|
||||
|
||||
@@ -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>
|
||||
|
||||
@@ -32,6 +32,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();
|
||||
|
||||
@@ -31,6 +31,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();
|
||||
|
||||
@@ -1,22 +1,34 @@
|
||||
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.nio.file.Paths;
|
||||
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 {
|
||||
@@ -31,8 +43,10 @@ public class ExampleServerR4IT {
|
||||
|
||||
static {
|
||||
HapiProperties.forceReload();
|
||||
HapiProperties.setProperty(HapiProperties.DATASOURCE_URL, "jdbc:derby:memory:dbr4;create=true");
|
||||
HapiProperties.setProperty(HapiProperties.FHIR_VERSION, "R4");
|
||||
HapiProperties.setProperty(HapiProperties.TEST_PORT, Integer.toString(PortUtil.findFreePort()));
|
||||
HapiProperties.setProperty(HapiProperties.SUBSCRIPTION_WEBSOCKET_ENABLED, "true");
|
||||
ourCtx = FhirContext.forR4();
|
||||
ourPort = HapiProperties.getTestPort();
|
||||
}
|
||||
@@ -50,6 +64,66 @@ public class ExampleServerR4IT {
|
||||
assertEquals(methodName, pt2.getName().get(0).getFamily());
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void testWebsocketSubscription() throws Exception {
|
||||
/*
|
||||
* Create subscription
|
||||
*/
|
||||
Subscription subscription = new Subscription();
|
||||
subscription.setReason("Monitor new neonatal function (note, age will be determined by the monitor)");
|
||||
subscription.setStatus(Subscription.SubscriptionStatus.REQUESTED);
|
||||
subscription.setCriteria("Observation?status=final");
|
||||
|
||||
Subscription.SubscriptionChannelComponent channel = new Subscription.SubscriptionChannelComponent();
|
||||
channel.setType(Subscription.SubscriptionChannelType.WEBSOCKET);
|
||||
channel.setPayload("application/json");
|
||||
subscription.setChannel(channel);
|
||||
|
||||
MethodOutcome methodOutcome = ourClient.create().resource(subscription).execute();
|
||||
IIdType mySubscriptionId = methodOutcome.getId();
|
||||
|
||||
// Wait for the subscription to be activated
|
||||
waitForSize(1, () -> ourClient.search().forResource(Subscription.class).where(Subscription.STATUS.exactly().code("active")).cacheControl(new CacheControlDirective().setNoCache(true)).returnBundle(Bundle.class).execute().getEntry().size());
|
||||
|
||||
/*
|
||||
* Attach websocket
|
||||
*/
|
||||
|
||||
WebSocketClient myWebSocketClient = new WebSocketClient();
|
||||
SocketImplementation mySocketImplementation = new SocketImplementation(mySubscriptionId.getIdPart(), EncodingEnum.JSON);
|
||||
|
||||
myWebSocketClient.start();
|
||||
URI echoUri = new URI("ws://localhost:" + ourPort + "/websocket");
|
||||
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();
|
||||
@@ -57,7 +131,13 @@ public class ExampleServerR4IT {
|
||||
|
||||
@BeforeClass
|
||||
public static void beforeClass() throws Exception {
|
||||
String path = Paths.get("").toAbsolutePath().toString();
|
||||
/*
|
||||
* 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);
|
||||
|
||||
@@ -68,6 +148,7 @@ public class ExampleServerR4IT {
|
||||
|
||||
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);
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user