tutorial - http support
This is a beginner tutorial about writing http-based network applications by using xSocket. If you need more help, don't hesitate to use xSocket's support forum.
http protocol support (xSocket extension)
xSocket supports writing http-based client and server applications. xSocket supports synchronous, blocking HTTP as well as asynchronous non blocking HTTP. An introduction to asynchronous, non-blocking HTTP programming can be found here.The http connection represents the key abstraction. It bases internally on a INonBlockingConnection instance to handle the tcp network communication. The integration of the http connection is highly optimized by reducing the overhead of layering.
A INonBlockingConnection can become a http connection at any time. This is true for the client-side as well as for the server-side.
.
To use xSocket-http please add the libraries xSocket-http-2.0-alpha-4.jar and xSocket-2.0-beta-2.jar to your classpath.
1. Client-side http connection
A new http client-side connection can be established in two ways. Either based on a existing tcp connection by wrapping it
INonBlockingConnection tcpCon = new NonBlockingConnection("www.gmx.com", 80);
IHttpClientEndpoint httpClientConnection = new HttpClientConnection(tcpCon);
... or in a direct way.
IHttpClientEndpoint httpClientConnection = new HttpClientConnection("www.gmx.com", 80);
The http client connection allows to send requests to the server and to receive responses from the server in a comfortable way. The http connection supports synchronous calls as well as a asynchronous, handler-based communication model. In contrast to the asynchronous approach, the current thread will be suspended within a synchronous call until the response header of the server-side has been received.
1.1 Synchronous client call
To perform a synchronous call a request object has to be created and the HttpClientConnection's call method has to be performed. xSocket supports a generic Request object as well as more comfortable, dedicated request objects such as PostRequest or GetRequest.To request object provides (Servlet API compatible) method to access header data. By sending the request xSocket adds the HTTP 1.1 mandatory request header parameters, if they are not present. By using the http connection, the host and port parameter within the request URL is optional. If such parameters are not present the remote host or remote port of the underlying connection will be used to add the http request header host field.
IHttpRequest request = new GetRequest("http://www.gmx.com/index.html");
// request = new GetRequest("/index.html"); would also be OK
// add request header data
request.setHeader("Accept-Encoding", "gzip,deflate");
// perform the request
IHttpResponse response = httpClientConnection.call(request);
// get repsonse header data
String contentType = response.getContentType();
//...
The call method sends the request to the server and returns a response object by receiving the response header from the server. The response object provides several convenience methods to retrieve the header data. The body data can be access by using the get body methods. The returned body data source object implements some convenience methods to read the body data such as read<dataType>().
//... BlockingBodyDataSource bodyDataSource = response.getBlockingBody(); int i = bodyDataSource.readInt(); //...
The data source object also implements the ReadableByteChannel interface of the java.nio package. If a InputStream is required instead of a ReadableByteChannel the java.nio.Channels.newInputStream(<readableChannel>) method can be used to wrap the ReadableByteChannel by an InputStream.
//... ReadableByteChannel bodyDataSource = response.getBlockingBody(); InputStream is = Channels.newInputStream(bodyDataSource); //...
By returning from the call method, it is not ensured, that the (complete) response body has been received. The method returns immediately after the complete response header has been received. For this reason xSocket provides two different getBody methods.
1.2 Retrieve the response body data in a blocking mode
A body handle which implements a blocking access manner can be retrieved by calling the getBlockingBody method. This method returns a BlockingBodyDataSource. By calling the data retrieve methods of the data source object, the method call blocks until enough body data is available. By retrieving more sufficient body data or by reaching the timeout, the method call returns (in the case of timeout by throwing a timeout exception).BlockingBodyDataSource bodyDataSource = response.getBlockingBody(); String data = bodyDataSource.readStringByLength(length); //...
1.3 Retrieve the body data in a non-blocking mode
In contrast to the blocking access behaviour, a NonBlockingBodyDataSource will return immediately (possible by throwing a BufferUnderflowException if the available data size is to small). A non blocking data source will be retrieved by calling the getNonBlockingBody method.
To get notifications by receiving more body data a IBodyDataHandler can be set at any time for a data source object.
NonBlockingBodyDataSource bodyDataSource = response.getNonBlockingBody(); //... IBodyDataHandler bodyHandler = new BodyToFileStreamer(file); response.getNonBlockingBody().setDataHandler(bodyHandler);
The data handler has to implement a call back method onData, which will be called each time by a worker thread when new data has been received. Using a data handler allows to implement a non blocking streaming of the body data. Especially for large bodies this can increase the efficiency significantly by avoiding blocking reads (which causes wasting resources by suspending threads).
class BodyToFileStreamer implements IBodyDataHandler {
private FileChannel fc = null;
private File file = null;
BodyToFileStreamer(File file) throws IOException {
this.file = file;
fc = new RandomAccessFile(file, "rw").getChannel();
}
public boolean onData(NonBlockingBodyDataSource bodyChannel) {
try {
int available = bodyChannel.available();
// data to transfer?
if (available > 0) {
bodyChannel.transferTo(fc, available);
// end of stream reached?
} else if (available == -1) {
fc.close();
}
} catch (IOException ioe) {
file.delete();
}
return true;
}
}
The body handler supports the Execution annotation. The Execution annotation allows to define if the body handler should be called in a MULTITREADED or NONTHREADED mode. The default mode is MULTITREADED. For more information about the execution mode see xSocket’s core tutorial.
class MyBodyHandler implements IBodyDataHandler {
@Execution(Execution.NONTHREADED) // could also be set on the class level
public boolean onData(NonBlockingBodyDataSource bodyDataSource) {
// perform some non I/O bound operations
// ...
return true;
}
}
Using data handler allows reading the body data in a very efficient way. But by performing a synchronous call, the current thread will be suspended until the response header has been received. To avoid this a asynchronous call could be performed.
1.4 Asynchronous client call
By performing a synchronous call a IHttpResponseHandler has to be passed over to handle the response. The response handler’s onResponse method will be called by a worker thread, if the response header is received. The handler have also to implement a onException method, which will be called if an exception occurs.
class MyResponseHandler implements IHttpResponseHandler {
public void onResponse(IHttpResponse response) throws IOException {
int status = response.getStatus();
// ...
}
public void onException(IOException ioe) {
//...
}
}
The request will be send by calling the send method. The method returns immediately.
IHttpResponseHandler responseHandler = new MyResponseHandler();
IHttpRequest request = new GetRequest("http://www.gmx.com/index.html");
httpClientConnection.send(request, responseHandler);
//...
The response handler supports the Execution and InvokeOn annotation. The InvokeOn annotation allows defining if the onResponse method should be called by receiving the message header (default) or after receiving the complete message inclusive the body data. The InvokeOn annotation has to be set on the class level as well as on the method level.
@Execution(Execution.NONTHREADED) // could be also set on the method level
class MyResponseHandler implements IHttpResponseHandler {
@InvokeOn(InvokeOn.MESSAGE_RECEIVED)
public void onResponse(IHttpResponse response) throws IOException {
NonBlockingBodyDataSource bodyDataSource = response.getNonBlockingBody();
//...
}
public void onException(IOException ioe) {
//...
}
}
The asynchronous call feature allows you to perform requests regarding to the HTTP pipelining specification. Here multiple http requests are written out to a single http connection without waiting for the corresponding responses.
1.5 Asynchronous client call by streaming the request body
To support a non blocking streaming of the request body send methods exist which consumes a RequestHeader instead of the complete Request object. This method return a BodyDataSink object which will be used to write the body data in a non blocking way. The data sink object provides convenience methods and implements the WritableByteChannel interface of the java.nio package as well as the Flushable interface of the java.io package.
Calling the send method doesn't send the header data to the server immediately. The header data will be written by the first data sink flush operation. In the same way the body data will only be send by flushing the data sink. Closing the data sink also performs a flush, internally.
If no length parameter will be passed over by calling the send method, the body data will be send in a chunked mode. To send data in the plain mode the body length field has to be passed over. The close method has also to be performed to finish the send procedure.
// create a response handler
IHttpResponseHandler hdl = new MyResponseHandler();
// get the file to transfer
FileChannel fc = new RandomAccessFile("test.txt", "rw").getChannel();
int bodyLength = (int) fc.size();
// create a new request header
IHttpRequestHeader header = new HttpRequestHeader("POST", "http://server:80/in", "text/plain");
// start sending in non-chunked mode
BodyDataSink bodyDataSink = httpClientConnection.send(header, bodyLength, hdl);
// use the transfer method to write the body data
bodyDataSink.transferFrom(fc);
// finish the send procedure
bodyDataSink.close();
The flush behaviour can be controlled by setting the flush mode (SYNC, ASYNC). For more information about flushing see xSocket’s core tutorial. By default the autoflush is activated. That means, each write operation performs a flush, implicitly (and the header will be written within the first write operation). The autoflush can be deactivated by calling the setAutoflush method.
IHttpResponseHandler responseHandler = new MyResponseHandler();
// create a new request header
IHttpRequestHeader header = new HttpRequestHeader("POST", "http://server:80/in", "text/plain");
header.setHeader("Accept-Encoding", "gzip,deflate");
// start sending in chunked mode
BodyDataSink bodyDataSink = httpClientConnection.send(header, responseHandler);
// set some data sink properties
bodyDataSink.setAutoflush(false); // deactivate the autoflush
bodyDataSink.setFlushmode(FlushMode.ASYNC); // override default flush mode 'SYNC'
// writing and flushing data
bodyDataSink.write(...);
bodyDataSink.flush();
//...
// finish the send procedure
bodyDataSink.close();
2. HttpClient
Often a higher level representation is desirable to perform client-side http requests. By using the HttpClient the http connection management will be handled automatically.This means creating, (re)using and deleting of the IHttpClientConnections will be done by the HttpClient. The HttpClient uses a connection pool, internally. To give control over the pooling behaviour the HttpClient implements the IConnectionPool interface. For more information about connection pooling see xSocket’s core tutorial.
Equals to the HttpClientConnection the HttpClient implements the IHttpClientEndpoint interface which defines the client-side call and send methods. The code to send requests is like the code described in the chapters above.
IHttpClientEndpoint httpClient = new HttpClient();
IHttpRequest request = new GetRequest("http://www.gmx.com/index.html");
// request = new GetRequest("/index.html"); wouldn't work here!
// add request header data
request.setHeader("Accept-Encoding", "gzip,deflate");
// perform the request
IHttpResponse response = httpClient.call(request);
// get response header data
String contentType = response.getContentType();
//...
httpClient.close();
After using the HttpClient it should be closed. By closing the HttpClient the internal pool watch dog thread will be terminated.
To call a https based URL an SSLContext object has to be passed over within the HttpClient constructor. The SSLContext object will be used to establish HTTPS connections.
// pass over the SSL context (here the JSE 1.6 getDefault method will be used)
HttpClient httpClient = new HttpClient(SSLContext.getDefault());
// make some configurations
httpClient.setMaxIdle(3); // configure the pooling behaviour
httpClient.setTreat302RedirectAs303(true); // redirect 302 post response by a get request
ConnectionUtils.registerMBean(httpClient); // register the http client's mbean
// create a request
String[] formParams = new String[] {"username=myname",
"password=I don't tell you"};
PostRequest request = new PostRequest("https://login.web.de/intern/login/", formParams);
// call it by following redirects
IHttpResponse response = httpClient.callFollowRedirects(request);
// get the redirected response
BlockingBodyDataSource bodyDataSource = response.getBlockingBody();
//...
As you see in the example above, the HttpClient provides additional call methods. By calling the callFollowRedirects or the sendFollowRedirects method the HttpClient follows the redirects and returns the redirected response. Such methods are not defined by the IHttpClientEndpoint interface and are exclusive to the HttpClient.
By registering the HttpClient's mbean you can get information about the HttpClient
3. Server-side http connection
3.1 Server handler
To handle incoming http request on the server side a server handler has to be implemented. The IHttpRequestHandler interface defines an onRequest method, which will be called, each time a new request is received by the server.
By calling the method the IHttpExchange will be passed over. This object encapsulates a HTTP request received and a response to be generated in one exchange. It provides methods to retrieve and forward the the request, and to send the response. The response will be send by calling the send(Response) method. An error response can also be send by using the sendError(...) method.
class MethodInfoHandler implements IHttpRequestHandler {
public void onRequest(IHttpExchange exchange) throws IOException {
IHttpRequest request = exchange.getRequest();
// ...
// create a response object
IHttpResponse resp = new HttpResponse("text/plain", "called method=" + req.getMethod());
// ... and send it back to the client
exchange.send(resp);
}
}
Equals the response object discussed above, the request object supports a getNonBlockingBody() method as well as a getBlockingBody() method. By using the getBlockingBody() method it has to be considered, that the body data retrieve methods blocks which causes that the current thread can be suspended. Retrieving the body by the getBlockingBody() method should never be used within the NONTHREADED mode.
Based on the server handler a server can be instantiated.
// creates the server by passing over the port number & the server handler IServer srv = new HttpServer(80, new MethodInfoHandler()); // run it srv.run(); // ... or run it by using the ConnectionUtils ConnectionUtils.start(srv); // returns after the server has been started
3.2 Message Forwarding (Proxy example)
Beside the send method, the IHttpExchange also supports a forward method. This method can be used to forward the request. For instance by implementing a proxy, the message can be forwarded by this method.
xSocket client-side und server-side support is highly integrated. On both sides, the same programming style and classes will be used. For this reason it is very easy to write http proxies or routers which typically receives, modifies and forwards http messages. xSocket implements a lot of optimizations to support such UseCases . For instance if a message is received and forwarded, xSocket will stream the body internally in a non-blocking way.
class ForwardHandler implements IHttpRequestHandler {
private String targetHost = null;
private int targetPort = -1;
ForwardHandler(String targetHost, int targetPort) {
this.targetHost = targetHost;
this.targetPort = targetPort;
}
@Execution(Execution.MULTITHREADED)
@InvokeOn(InvokeOn.HEADER_RECEIVED)
public void onRequest(IHttpExchange exchange) throws IOException {
IHttpRequest req = exchange.getRequest();
// remove the named hop-by-hop headers
req.removeHopByHopHeaders("CONNECTION", "PROXY-AUTHORIZATION",
"TRAILER", "UPGRADE");
// perform further proxy issues (via header, cache, ...)
// ...
// reset address (Host header will be update automatically)
req.updateTargetURI(targetHost, targetPort);
// .. and forward the request
try {
exchange.forward(req, new ReverseHandler(exchange));
} catch (ConnectException ce) {
exchange.sendError(502, ce.getMessage());
}
}
}
The ForwardHandler of the proxy example receives messages, modifies and forwards it. By donig this the Hop-by-hopheaders have to be removed. The Hop-by-hop headers are only valid for the for a single transport-level connection. The Hop-by-hop headers can be removed by calling the removeHophByHopHeaders.The remove method considers also the specific algorithm of the connection header. The some standard hop-by-hop headers will be handled automatically by the xSocket. That means If the response contains a connection: close header, the connection will be closed, automatically. xSocket handle the hop-by-hop heades as follows
| hop-by-hop header | client-side (server response) | server-side (client request) |
|---|---|---|
| Connection | auto handled | auto handled |
| Keep-Alive | auto handled | |
| Proxy-Authenticate | not handled | |
| Proxy-Authorization | not handled | |
| Proxy-Connection | not handled | |
| TE | not handled | |
| Trailer | not handled | not handled |
| Transfer-Encoding | auto handled | auto handled |
| Upgrade | not handled | not handled |
If a chunked message contains trailers defined by the Trailer header, these trailers will be added to Response object’s header list after reading it.
The ReverseHandler implements the IHttpResponseHandler interface as well as the IHttpResponseTimeoutHandler interface. The onResponseTimeout method will be called, if the response timeout is reached. The response timeout can be set on the HttpClient and on the HttpClientConnection class.
class ReverseHandler implements IHttpResponseHandler, IHttpResponseTimeoutHandler {
private IHttpExchange exchange = null;
public ReverseHandler(IHttpExchange exchange) {
this.exchange = exchange;
}
@Execution(Execution.NONTHREADED)
@InvokeOn(InvokeOn.HEADER_RECEIVED)
public void onResponse(IHttpResponse resp) throws IOException {
// handle proxy issues (hop-by-hop headers, ...)
// ...
// return the response
exchange.send(resp);
}
@Execution(Execution.NONTHREADED)
public void onException(IOException ioe) {
exchange.sendError(500, ioe.toString());
}
@Execution(Execution.NONTHREADED)
public void onException(SocketTimeoutException stoe) {
exchange.sendError(504, stoe.toString());
}
}
The IhttpExchange forward method uses internally a HttpClient instance to forward the request. The forward HttpClient be set manually by calling the setForwardHttpClient() method.
// create the proxy server HttpServer proxy = new HttpServer(listenport, new ForwardHandler()); // define a custom configured forward HttpClient HttpClient httpClient = new HttpClient(); httpClient.setWorkerpool(proxy.getWorkerpool()); httpClient.setResponseTimeoutSec(receiveTimeoutSec); // .. set it proxy.setForwardHttpClient(httpClient); // and run the proxy proxy.run();
The forward method will also be used to forward the request (locally) within a request handler chain.
3.3 Handler chaining
By using a RequestHandlerChain handlers can be chained. The RequestHandlerChain implements the IHttpRequestHandler interface and will be handled as a regular IHttpRequestHandler implementation.
// define a chain RequestHandlerChain chain = new RequestHandlerChain(); chain.addLast(new LogFilter()); chain.addLast(new ForwardHandler()); HttpServer proxy = new HttpServer(listenport, chain); proxy.run();
The LogFilter defined here creates a copy of the request and response message and prints it out. The message will be forwarded locally by calling the forward method. In this case the next RequestHandler of the chain will be called. Here the ForwardHandler of the example above will be called.
xSocket detects automatically, if the HttpRequest to forward is a remote or a local request. If the request is a local request, the next handler of the chain will be called. If the request is a remote, external request, the external resource will called instead of the next handler of the chain. By non-forwarding the request, the handling chain will be interrupted.
By using the forwarded method based on the request header, the body can be stream without buffering the total message.
The LogFilter here also uses the xSocket AbstractBodyForwarder convenience class to forward be message body.
class LogFilter implements IHttpRequestHandler {
public void onRequest(final IHttpExchange exchange) throws IOException {
IHttpRequest req = exchange.getRequest();
// is body less request?
if (!req.hasBody()) {
System.out.println(req.getRequestHeader().toString());
exchange.forward(req, new ResponseHandler(exchange));
// .. no, the request do have a body
} else {
// get the request header and body
final IHttpRequestHeader header = req.getRequestHeader();
NonBlockingBodyDataSource bodyDataSource = req.getNonBlockingBody();
// create a log buffer
final List<ByteBuffer> logBuffer = new ArrayList<ByteBuffer>();
// forward the request (header)
final BodyDataSink bodyDataSink = exchange.forward(req.getRequestHeader(), new ResponseHandler(exchange));
// define a forward handler
AbstractBodyForwarder bodyForwardHandler = new AbstractBodyForwarder(bodyDataSource, bodyDataSink) {
@Override
public void onData(NonBlockingBodyDataSource source, BodyDataSink sink) throws IOException {
ByteBuffer[] bufs = source.readByteBufferByLength(source.available());
for (ByteBuffer byteBuffer : bufs) {
logBuffer.add(byteBuffer.duplicate());
}
sink.write(bufs);
}
@Override
public void onComplete() {
System.out.println(header.toString());
try {
System.out.println(header.toString() +
DataConverter.toString(logBuffer, header.getCharacterEncoding()));
} catch (Exception e) {
System.out.println("<body not printable>");
}
}
};
// an set it
bodyDataSource.setDataHandler(bodyForwardHandler);
}
}
}
Similar to the request handler the response handler also uses the AbstractBodyForwarder class.
class ResponseHandler implements IHttpResponseHandler {
private IHttpExchange exchange = null;
ResponseHandler(IHttpExchange exchange) {
this.exchange = exchange;
}
public void onResponse(IHttpResponse response) throws IOException {
// is body less response?
if (!response.hasBody()) {
try {
System.out.println(response.getResponseHeader());
} catch (Exception ignore) { }
exchange.send(response);
// ... no, it has a body
} else {
// get the response header and body
final IHttpResponseHeader header = response.getResponseHeader();
NonBlockingBodyDataSource bodyDataSource = req.getNonBlockingBody();
// create a log buffer
final List<ByteBuffer> logBuffer = new ArrayList<ByteBuffer>();
// send the response (header)
final BodyDataSink bodyDataSink = exchange.send(response.getResponseHeader());
// define a forward handler
AbstractBodyForwarder bodyForwardHandler = new AbstractBodyForwarder(bodyDataSource, bodyDataSink) {
@Override
public void onData(NonBlockingBodyDataSource source, BodyDataSink sink) throws IOException {
ByteBuffer[] bufs = source.readByteBufferByLength(source.available());
for (ByteBuffer byteBuffer : bufs) {
logBuffer.add(byteBuffer.duplicate());
}
sink.write(bufs);
}
@Override
public void onComplete() {
System.out.println(header.toString());
try {
System.out.println(header.toString() +
DataConverter.toString(logBuffer, header.getCharacterEncoding()));
} catch (Exception e) {
System.out.println("<body not printable>");
}
}
};
// an set it
bodyDataSource.setDataHandler(bodyForwardHandler);
}
}
public void onException(IOException ioe) {
exchange.sendError(500);
}
}
In the example below a AuthHandler filter is used which implements the http basic authorization. Each request first enters the AuthHandler, which validates the authorization. If the authorization fails, the AuthHandler will send a error message. If the authorization passes, the next handler of the chain, the FileServiceHandler will be called by forwarding the request.
RequestHandlerChain chain = new RequestHandlerChain();
chain.addLast(new AuthHandler());
chain.addLast(new FileServiceHandler("C:\\temp", true));
IServer server = new HttpServer(port, chain);
server.run();
class AuthFilter implements IHttpRequestHandler {
private Authenticator authenticator = new Authenticator();
public void onRequest(final IHttpExchange exchange) throws IOException {
IHttpRequest request = exchange.getRequest();
String authorization = request.getHeader("Authorization");
if (authorization != null) {
String[] s = authorization.split(" ");
if (!s[0].equalsIgnoreCase("BASIC")) {
exchange.sendError(401);
return;
}
String decoded = new String(Base64.decodeBase64(s[1].getBytes()));
String[] userPasswordPair = decoded.split(":");
String authtoken = authenticator.login(userPasswordPair[0], userPasswordPair[1]);
IHttpResponseHandler respHdl = new IHttpResponseHandler() {
public void onResponse(IHttpResponse response) throws IOException {
exchange.send(response);
}
public void onException(IOException ioe) {
exchange.sendError(500);
}
};
exchange.forward(exchange.getRequest(), respHdl);
return;
}
String authentication = request.getHeader("Authentication");
if (authentication == null) {
HttpResponse resp = new HttpResponse(401);
resp.setHeader("WWW-Authenticate", "basic");
exchange.send(resp);
}
}
The xSocket-http built-in FileServiceHandler returns the requested file based on the configured base path and the requested URI resource.
3.4 Routing (URL patterns support)
Analogous to the Servlet approach specific request handlers can be
assigned via a url-pattern to a set of URLs. The
url-pattern syntax is equals to the Servlet approach. A request handler
will be assigned to a url by using a Context.
Typically, this approach is required, if static resources have to be supported as well as dynamic resources.
Context rootCtx = new Context("");
rootCtx.addHandler("/service/*", new TimerHandler());
rootCtx.addHandler("/*", new FileServiceHandler("C:\\apps\timer\files));
IServer server = new HttpServer(port, rootCtx);
server.run();
class CometForeverFrameHandler implements IHttpRequestHandler {
private final Timer timer = new Timer("timer", true);
public void onRequest(final IHttpExchange exchange) throws IOException {
IHttpRequest req = exchange.getRequest();
// handle the time request
if (req.getRequestURI().endsWith("/time")) {
// write the message header by retrieving the body handle
IHttpResponseHeader respHdr = new HttpResponseHeader(200, "text/html");
final BodyDataSink outChannel = exchange.send(respHdr);
// timer task definition
TimerTask timerTask = new TimerTask() {
public void run() {
try {
String script = "<script>\r\n" +
" parent.printTime(\"" + new Date() + "\");\r\n" +
"</script>";
outChannel.write(script);
} catch (IOException ioe) {
cancel();
try {
outChannel.close();
} catch (IOException ignore) { }
}
}
};
// start the timer task
timer.schedule(timerTask, 0, 1000);
}
}
}
The example above defines the FileServiceHandler as default
handler (namespace /*). The TimeHandler is
assigned to the namespace /service/*, which means if the
requested URI matches with this namespace, the TimeHandler will be used
to serve the request.
3.5 Asynchronous processing - a multicast example
Equals to the client-side the server-side supports asynchronous processing of the http messages. This allows to write server push applications, which sends (body) data to client.
The following multicast example shows a simple server-side implementation that realizes a multicast bus. The MulticastServerHandler reads the incoming body data of the connections, and forwards the data packets to all open http connections in a asynchronous way. In this example each body data packet has to start with an application-level int length field.
The group management of the open connections will be done here by a MulticastService.
class MulticastService {
private final Map<String, WritableByteChannel> peers = new HashMap<String, WritableByteChannel>();
synchronized int getCountPeers() {
return peers.size();
}
synchronized void registerPeer(String id, WritableByteChannel channel) {
peers.put(id, channel);
}
synchronized void deregisterPeer(String id) {
peers.remove(id);
}
synchronized void sendToPeers(ByteBuffer[] data) {
for (Entry<String, WritableByteChannel> peer : peers.entrySet()) {
try {
for (ByteBuffer buf : data) {
peer.getValue().write(buf);
buf.flip();
}
} catch (IOException ioe) {
deregisterPeer(peer.getKey());
}
}
}
}
The IRequestHandler implementation receives new incoming http request. Each request will be answered by a proper http response. The request and response body is read and written in an asynchronous non-blocking way. The example code makes also use of the Execution annotations. For more information about execution and optimizations see xSocket’s core tutorial.
As you see in the example code, the server handler also implements the IHttpDisconnectHandler interface. The onDisconnect(...) method, defined by this interface will be called if a http connection is terminated. xSocket also supports a IHttpConnectHandler interface, which will be called if a new http connection is established.
class MulticastServerHandler implements IHttpRequestHandler, IHttpDisconnectHandler {
private MulticastService multicastService = null;
public MulticastServerHandler(MulticastService multicastService) {
this.multicastService = multicastService;
}
@Execution(Execution.NONTHREADED)
public void onRequest(final IHttpExchange exchange) throws IOException {
IHttpRequest request = exchange.getRequest();
// retrieve request body
NonBlockingBodyDataSource requestBody = request.getNonBlockingBody();
// prepare and send response
IHttpResponseHeader responseHeader = new HttpResponseHeader(200);
BodyDataSink bodyDataSink = exchange.send(responseHeader);
multicastService.registerPeer(exchange.getConnection().getId(), bodyDataSink);
IBodyDataHandler bodyHandler = new IBodyDataHandler() {
public boolean onData(NonBlockingBodyDataSource bodyDataSource)
throws BufferUnderflowException {
try {
// each message start with a packet size field
HttpUtils.validateSufficientDatasizeByIntLengthField(bodyDataSource, false);
int available = bodyDataSource.available();
ByteBuffer[] availableData = bodyDataSource.readByteBufferByLength(available);
multicastService.sendToPeers(availableData);
return true;
} catch (IOException ioe) {
multicastService.deregisterPeer(exchange.getConnection().getId());
}
return true;
}
};
requestBody.setDataHandler(bodyHandler);
}
@Execution(Execution.NONTHREADED)
public boolean onDisconnect(IHttpConnection httpConnection) throws IOException {
multicastService.deregisterPeer(httpConnection.getId());
return true;
}
}
By default the server handler is instance-scoped. That means, the same handler instance will be used for each new incoming http connection. For this reason the MulticastService object above represents a singleton object .
A handler becomes connection-scoped by implementing the IConnectionScoped interface. This interface requires a clone method, which will be used to create a dedicated instance of the given handler for each new connection.
4 Using xSocket-http together with Spring
For basic information about xSocket and Spring see xSocket’s core tutorial. The example here shows how to use xSocket http's RequestHandlerChain, Context and HttpClient together with Spring.
By using a RequestHandlerChain the request handler can be set by using Spring's constructor Injection.
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
<bean id="server" class="org.xsocket.connection.http.server.HttpServer" scope="singleton"
init-method="start" destroy-method="close">
<constructor-arg value="8080"/>
<constructor-arg ref="chain"/>
</bean>
<bean id="chain" class="org.xsocket.connection.http.RequestHandlerChain"
scope="prototype">
<constructor-arg>
<list>
<ref bean="authFilter"/>
<ref bean="handler"/>
</list>
</constructor-arg>
</bean>
<bean id="authFilter" class="mynamespace.MyAuthFilter" scope="prototype"/>
<bean id="handler" class="mynamespace.MyHandler" scope="prototype"/>
</beans>
The constructor injection can also used to create a Context.
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
<bean id="server" class="org.xsocket.connection.http.server.HttpServer" scope="singleton"
init-method="start" destroy-method="close">
<constructor-arg value="8080"/>
<constructor-arg ref="context"/>
</bean>
<bean id="context" class="org.xsocket.connection.http.server.Context" scope="prototype">
<constructor-arg value=""/>
<constructor-arg>
<map>
<entry key="/info/*">
<ref bean="infoHandler"/>
</entry>
<entry key="/files/*">
<ref bean="fileServiceHandler"/>
</entry>
</map>
</constructor-arg>
</bean>
<bean id="infoHandler" class="mynamespace.MyHandler" scope="prototype"/>
<bean id="fileServiceHandler" class="org.xsocket.connection.http.server.FileServiceHandler"
scope="prototype">
<constructor-arg index="0" value="C:\temp"/>
<constructor-arg index="1" value="true"/>
</bean>
</beans>
To create and configure a HttpClient see the example below.
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
<bean id="httpClient" class="org.xsocket.connection.http.client.HttpClient"
scope="prototype">
<property name="maxIdle">
<value>30</value>
</property>
<property name="responseTimeoutSec">
<value>120</value>
</property>
</bean>
</beans>