Added sample Webapp to use an EvebtBus on top of Socket.IO · tadglines/Socket.IO-Java@490a3a4 · GitHub
Skip to content

Commit 490a3a4

Browse files
author
Mathieu Carbou
committed
Added sample Webapp to use an EvebtBus on top of Socket.IO
1 parent aee8aa7 commit 490a3a4

10 files changed

Lines changed: 749 additions & 0 deletions

File tree

‎pom.xml‎

Lines changed: 12 additions & 0 deletions

‎samples/eventbus/pom.xml‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
3+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
4+
5+
<modelVersion>4.0.0</modelVersion>
6+
7+
<parent>
8+
<groupId>com.glines.socketio.sample</groupId>
9+
<artifactId>socketio-sample</artifactId>
10+
<version>0.1-SNAPSHOT</version>
11+
<relativePath>..</relativePath>
12+
</parent>
13+
14+
<artifactId>socketio-sample-eventbus</artifactId>
15+
<packaging>war</packaging>
16+
17+
<name>Socket.IO Sample EventBus</name>
18+
<description>${project.name}</description>
19+
20+
<dependencies>
21+
<dependency>
22+
<groupId>com.glines.socketio</groupId>
23+
<artifactId>socketio-core</artifactId>
24+
</dependency>
25+
<dependency>
26+
<groupId>org.codehaus.jettison</groupId>
27+
<artifactId>jettison</artifactId>
28+
</dependency>
29+
30+
<dependency>
31+
<groupId>org.testatoo.container</groupId>
32+
<artifactId>testatoo-container-jetty</artifactId>
33+
<scope>provided</scope>
34+
</dependency>
35+
</dependencies>
36+
37+
</project>

‎samples/eventbus/src/main/java/com/glines/socketio/sample/eventbus/EventBusServlet.java‎

Lines changed: 278 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,278 @@
1+
/**
2+
* The MIT License
3+
* Copyright (c) 2010 Tad Glines
4+
*
5+
* Permission is hereby granted, free of charge, to any person obtaining a copy
6+
* of this software and associated documentation files (the "Software"), to deal
7+
* in the Software without restriction, including without limitation the rights
8+
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9+
* copies of the Software, and to permit persons to whom the Software is
10+
* furnished to do so, subject to the following conditions:
11+
*
12+
* The above copyright notice and this permission notice shall be included in
13+
* all copies or substantial portions of the Software.
14+
*
15+
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16+
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17+
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18+
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19+
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20+
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
21+
* THE SOFTWARE.
22+
*/
23+
package com.glines.socketio.sample.eventbus;
24+
25+
import com.glines.socketio.common.DisconnectReason;
26+
import com.glines.socketio.common.SocketIOException;
27+
import com.glines.socketio.server.SocketIOInbound;
28+
import com.glines.socketio.server.SocketIOOutbound;
29+
import com.glines.socketio.server.SocketIOServlet;
30+
import org.codehaus.jettison.json.JSONArray;
31+
import org.codehaus.jettison.json.JSONException;
32+
import org.codehaus.jettison.json.JSONObject;
33+
import org.eclipse.jetty.util.log.JavaUtilLog;
34+
import org.eclipse.jetty.util.log.Log;
35+
36+
import javax.servlet.http.HttpServletRequest;
37+
import java.io.IOException;
38+
import java.util.concurrent.ConcurrentHashMap;
39+
import java.util.concurrent.ConcurrentMap;
40+
import java.util.logging.Level;
41+
import java.util.logging.LogManager;
42+
import java.util.logging.Logger;
43+
44+
public class EventBusServlet extends SocketIOServlet {
45+
46+
static {
47+
try {
48+
Log.setLog(new JavaUtilLog());
49+
LogManager.getLogManager().reset();
50+
LogManager.getLogManager().readConfiguration(Thread.currentThread().getContextClassLoader().getResourceAsStream("logging.properties"));
51+
} catch (IOException e) {
52+
throw new RuntimeException(e.getMessage(), e);
53+
}
54+
}
55+
56+
private static final Logger LOGGER = Logger.getLogger(EventBusServlet.class.getName());
57+
58+
private final ConcurrentMap<String, Endpoints> subscriptions = new ConcurrentHashMap<String, Endpoints>();
59+
60+
@Override
61+
protected SocketIOInbound doSocketIOConnect(HttpServletRequest request) {
62+
return new Endpoint(request.getSession().getId(), request.getRemoteHost(), request.getRemotePort());
63+
}
64+
65+
private final class Endpoint implements SocketIOInbound {
66+
67+
private final String remoteHost;
68+
private final int remotePort;
69+
private final String id;
70+
71+
private volatile SocketIOOutbound outbound;
72+
73+
private Endpoint(String id, String remoteHost, int remotePort) {
74+
this.remoteHost = remoteHost;
75+
this.remotePort = remotePort;
76+
this.id = id;
77+
}
78+
79+
String getId() {
80+
return id;
81+
}
82+
83+
@Override
84+
public String toString() {
85+
return "Endpoint " + id + " (" + remoteHost + ":" + remotePort + ")";
86+
}
87+
88+
@Override
89+
public int hashCode() {
90+
return 31 * id.hashCode();
91+
}
92+
93+
@Override
94+
public boolean equals(Object o) {
95+
if (this == o) return true;
96+
if (o == null || getClass() != o.getClass()) return false;
97+
Endpoint endpoint = (Endpoint) o;
98+
return id.equals(endpoint.id);
99+
}
100+
101+
@Override
102+
public void onConnect(SocketIOOutbound outbound) {
103+
if (LOGGER.isLoggable(Level.FINE))
104+
LOGGER.fine(this + " connected.");
105+
this.outbound = outbound;
106+
try {
107+
send(new JSONObject().put("type", MessageType.ACK));
108+
} catch (JSONException e) {
109+
throw new RuntimeException(e.getMessage(), e);
110+
}
111+
}
112+
113+
@Override
114+
public void onDisconnect(DisconnectReason reason, String errorMessage) {
115+
if (LOGGER.isLoggable(Level.FINE))
116+
LOGGER.log(Level.FINE, this + " disconnected.");
117+
this.outbound = null;
118+
for (Endpoints ee : subscriptions.values())
119+
ee.remove(this);
120+
}
121+
122+
@Override
123+
public void onMessage(int messageType, String message) {
124+
if (outbound == null)
125+
throw new NullPointerException();
126+
if (LOGGER.isLoggable(Level.FINE))
127+
LOGGER.fine(this + " received message: " + message);
128+
try {
129+
JSONArray array = new JSONArray(message);
130+
for (int i = 0; i < array.length() && outbound != null; i++) {
131+
JSONObject json = array.getJSONObject(i);
132+
MessageType type = MessageType.valueOf(json.getInt("type"));
133+
switch (type) {
134+
case SUBSCRIBE: {
135+
String topic = json.getString("topic");
136+
if (LOGGER.isLoggable(Level.FINE))
137+
LOGGER.log(Level.FINE, this + " subscribes to topic: " + topic);
138+
subscriptions.putIfAbsent(topic, new Endpoints(topic));
139+
subscriptions.get(topic).add(this);
140+
break;
141+
}
142+
case UNSUBSCRIBE: {
143+
String topic = json.getString("topic");
144+
if (LOGGER.isLoggable(Level.FINE))
145+
LOGGER.log(Level.FINE, this + " unsubscribes from topic: " + topic);
146+
Endpoints ee = subscriptions.get(topic);
147+
if (ee != null)
148+
ee.remove(this);
149+
return;
150+
}
151+
case PUBLISH: {
152+
String topic = json.getString("topic");
153+
String data = json.getString("data");
154+
if (LOGGER.isLoggable(Level.FINE))
155+
LOGGER.log(Level.FINE, this + " publishes to topic " + topic + " message: " + data);
156+
Endpoints ee = subscriptions.get(topic);
157+
if (ee != null)
158+
ee.fire(topic, data);
159+
break;
160+
}
161+
case CLOSE: {
162+
close();
163+
break;
164+
}
165+
default: {
166+
close();
167+
throw new IllegalArgumentException("Illegal message: " + message);
168+
}
169+
}
170+
}
171+
} catch (JSONException e) {
172+
throw new RuntimeException(e.getMessage(), e);
173+
}
174+
}
175+
176+
void close() {
177+
if (LOGGER.isLoggable(Level.FINE))
178+
LOGGER.log(Level.FINE, this + " closing.");
179+
if (outbound != null) {
180+
outbound.close();
181+
this.outbound = null;
182+
}
183+
}
184+
185+
void send(JSONObject data) {
186+
if (outbound != null) {
187+
String str = data.toString();
188+
if (LOGGER.isLoggable(Level.FINE))
189+
LOGGER.fine("Sending to " + this + " message: " + str);
190+
try {
191+
outbound.sendMessage(str);
192+
} catch (SocketIOException e) {
193+
LOGGER.log(Level.SEVERE, "Error sending message to " + this + " => disconnecting. Error: " + e.getMessage(), e);
194+
close();
195+
}
196+
}
197+
}
198+
}
199+
200+
private final class Endpoints {
201+
202+
final String topic;
203+
final ConcurrentMap<String, Endpoint> endpoints = new ConcurrentHashMap<String, Endpoint>();
204+
205+
Endpoints(String topic) {
206+
this.topic = topic;
207+
}
208+
209+
@Override
210+
public String toString() {
211+
return "Endpoints for " + topic + ": " + endpoints.size();
212+
}
213+
214+
void add(Endpoint endpoint) {
215+
if (LOGGER.isLoggable(Level.FINE))
216+
LOGGER.log(Level.FINE, "Subscribing " + endpoint + " to " + this);
217+
Endpoint old = endpoints.get(endpoint.getId());
218+
if (old == null) {
219+
endpoints.putIfAbsent(endpoint.getId(), endpoint);
220+
} else {
221+
endpoints.replace(endpoint.getId(), old, endpoint);
222+
}
223+
if (old != null)
224+
old.close();
225+
}
226+
227+
void remove(Endpoint endpoint) {
228+
if (endpoints.remove(endpoint.getId(), endpoint)) {
229+
if (LOGGER.isLoggable(Level.FINE))
230+
LOGGER.log(Level.FINE, endpoint + " unsubscribed from " + this);
231+
if (endpoints.isEmpty()) {
232+
subscriptions.remove(topic, this);
233+
if (LOGGER.isLoggable(Level.FINE))
234+
LOGGER.log(Level.FINE, this + " removed from subscriptions.");
235+
}
236+
}
237+
}
238+
239+
void fire(String topic, String data) {
240+
for (Endpoint endpoint : endpoints.values())
241+
try {
242+
endpoint.send(new JSONObject().put("type", MessageType.PUBLISH).put("topic", topic).put("data", data));
243+
} catch (JSONException e) {
244+
throw new RuntimeException(e.getMessage(), e);
245+
}
246+
}
247+
}
248+
249+
private static enum MessageType {
250+
251+
ACK(4),
252+
CLOSE(5),
253+
SUBSCRIBE(1),
254+
UNSUBSCRIBE(2),
255+
PUBLISH(3),
256+
UNKNOWN(0);
257+
258+
final int code;
259+
260+
MessageType(int code) {
261+
this.code = code;
262+
}
263+
264+
265+
static MessageType valueOf(int code) {
266+
for (MessageType type : values())
267+
if (type.code == code)
268+
return type;
269+
return UNKNOWN;
270+
}
271+
272+
@Override
273+
public String toString() {
274+
return "" + code;
275+
}
276+
}
277+
278+
}

‎samples/eventbus/src/main/resources/logging.properties‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
handlers=java.util.logging.ConsoleHandler
2+
java.util.logging.ConsoleHandler.level=ALL
3+
com.glines.socketio.sample.eventbus.EventBusServlet.level=ALL
4+
.level=INFO

‎samples/eventbus/src/main/webapp/WEB-INF/web.xml‎

Lines changed: 48 additions & 0 deletions

0 commit comments

Comments
 (0)