Friday, 17 October 2014

WebSockets with Spring 4 SockJS and STOMP

Introduction



RFC 6455, the WebSocket protocol defines the capability for the web applications to be two-way, full-duplex  between client and server for communication. WebSocket helps to make the web to be more interactive including Java applets, XMLHttpRequest, server-sent events, and others. Spring 4 has introduced spring-websocket compatible with the Java WebSocket API.

WebSocket programming requires fallback options as a number of browsers don't support WebSocket (like IE10 and above supports WebSocket) and some restrictive proxies may be configured in ways that either preclude the attempt to do HTTP upgrade or otherwise break connection after some time because it has remained opened for too long. Spring Framework provides such transparent fallback options based on the SockJS protocol.

WebSocket application might use a single URL only for the initial HTTP handshake. All messages thereafter share and flow on the same TCP CONNECTION. This points to an entirely different, asynchronous, event-driven, messaging architecture


When to use WebSocket

The best fit for WebSocket is in web applications where client and server need to exchange events at high frequency and at low latency. Prime candidates include but are not limited to applications in finance, games, collaboration, and others. Such applications are both very sensitive to time delays and also need to exchange a wide variety of messages at high frequency.

Spring WebSocket components

Spring WebSocket implementation has some core components -

WebSocket Handler

A WebSocket handler is a class that will receive requests from the websocket client.


@Configuration
@EnableScheduling
@ComponentScan("org.springframework.samples")
@EnableWebSocketMessageBroker
public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer {


@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
registry.addEndpoint("/<some_name>").withSockJS();
}

}

Handshake Interceptor(Optional)

The websocket handshake interceptor is used to define and specify a class that intercepts the initial websocket handshake. Interceptors are purely optional.

public class HandshakeInterceptor extends HttpSessionHandshakeInterceptor{

@Override
public boolean beforeHandshake(ServerHttpRequest request,
ServerHttpResponse response, WebSocketHandler wsHandler,
Map<String, Object> attributes) throws Exception {

return super.beforeHandshake(request, response, wsHandler, attributes);
}

@Override
public void afterHandshake(ServerHttpRequest request,
ServerHttpResponse response, WebSocketHandler wsHandler,
Exception ex) {

super.afterHandshake(request, response, wsHandler, ex);
}

}

Enable SockJS

Enabling SockJS at server side is easy with configuration 

@Configuration
@EnableScheduling
@ComponentScan("org.springframework.samples")
@EnableWebSocketMessageBroker
public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer {


@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
registry.addEndpoint("/<some_name>").withSockJS();
}

}

STOMP


STOMP is the Simple (or Streaming) Text Orientated Messaging Protocol. STOMP provides an interoperable wire format so that STOMP clients can communicate with any STOMP message BROKERto provide easy and widespread messaging interoperability among many languages, platforms and BROKERS.

Spring  supports STOMP over WebSocket through the spring-messaging and spring-websocket modules. 

Here is an example of configuring a STOMP WebSocket endpoint with SockJS fallback options. The endpoint is available for clients to connect to at URL path /<app_name>/<end_point>:

@Configuration
@EnableScheduling
@ComponentScan("org.springframework.samples")
@EnableWebSocketMessageBroker
public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer {


@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
registry.addEndpoint("/<end_point>").withSockJS();
}

@Override
public void configureMessageBroker(MessageBrokerRegistry registry) {
registry.enableSimpleBroker("/queue/", "/topic/");
registry.setApplicationDestinationPrefixes("/<app_name>");
}

}

These are the benefits for an application from using STOMP over WebSocket:

  • Standard message format
  • Application-level protocol with support for common messaging patterns
  • Client-side support, e.g. stomp.js, msgs.js
  • The ability to interpret, route, and process messages on both client and server-side
  • The option to plug a message BROKER like RabbitMQ, ActiveMQ, many others to broadcast messages 

Sending Messages

It is also possible to send a message to user destinations from any application component by injecting the SimpMessageTemplate.
@Service
public class TradeServiceImpl implements TradeService {

private final SimpMessageTemplate messagingTemplate;

@Autowired
public TradeServiceImpl(SimpMessageTemplate messagingTemplate) {
this.messagingTemplate = messagingTemplate;
}

// ...

public void afterTradeExecuted(Trade trade) {
this.messagingTemplate.convertAndSendToUser(
trade.getUserName(), "<some_end_point>", trade.getResult());
}

}

SockJS

SockJS has the capability to use a WebSocket API but fall back to non-WebSocket alternatives when necessary at runtime without the need to change application code. As of writing this article only the following browsers supports WebSocket
  • IE 10
  • FireFox 6
  • Crome 4
  • Safari 5
  • Opera 12.10
To use SockJS we need sock.js at client side and SockJS server side library at server side.

WebSocket SockJS Client

SockJS client supports Ajax/XHR streaming in IE 8, 9 via Microsoft’s XDomainRequest. That works across domains but does not support sending cookies.
Java config to set SockJS client library - 
@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/<some_relative_url>").withSockJS()
                .setClientLibraryUrl("http://<host>:<post>/<myapp>/js/sockjs-client.js");
    }

    // ...


}

On the browser side, a client might connect as follows using stomp.js and the sockjs-client

var socket = new SockJS("/spring-websocket-portfolio/portfolio");
var stompClient = Stomp.over(socket);

stompClient.connect({}, function(frame) {

}

References:
http://docs.spring.io/spring/docs/current/spring-framework-reference/html/websocket.html
Working Code Example: 
https://github.com/badalb/spring-websocket-portfolio.git

Thursday, 16 October 2014

Distributed Session Management and Session Clustering

Distributed Session Management and Session Clustering

Introduction

Http is a stateless protocol, if there is a need to maintain the conversational state, session tracking is needed. For example, in a shopping cart application users keep on adding items into their cart using multiple requests. For every request, server should identify in which client’s cart the item is to be added. The standard ways to maintain http sessions are –
  •            User authorization
  •        Hidden fields
  •        URL rewriting
  •        Cookies
  •        Session tracking API

In a web based production environment we use load balancers to balance load across multiple servers. Managing sessions for such load balanced distributed environment is crucial specially to handle failovers.
We will be discussing some of the possible options to address the distributed session management and session clustering to overcome failovers. For simplicity we will be considering Apache web server and Tomcat as the app server only.

Load Balancing and Sticky Sessions

Sticky session is work on individual sessions. No session is shared with other tomcat servers. Server generated the session will keep that information of session, else no one know about that information. This can be done by Load balancer and mod_jk connectors at Apache level. If the underlying node is down sticky session don’t help, it only helps in load balancing.
To learn more about sticky session and load balancer setup we can the link below [http://www.ramkitech.com/2012/10/tomcat-clustering-series-part-2-session.html]

In Memory Session Broadcasting/Multicasting

Tomcat supports native in memory session clustering to handle load balancing, high performance and high availability. In this approach all the nodes participate in the cluster and every time a session is created the session information is shared to all the tomcat instances in the cluster.
To enable in memory session clustering in tomcat we need to configure a cluster in server.xml file available in ${CATALINA_HOME}/conf directory. The configuration snippet below will create a cluster in tomcat
<Engine name="Catalina" defaultHost="<some_host_name>" jvmRoute="<worker_name>”
<Cluster className="org.apache.catalina.ha.tcp.SimpleTcpCluster" channelSendOptions="8">
<Manager className="org.apache.catalina.ha.session.DeltaManager" expireSessionsOnShutdown="false" notifyListenersOnReplication="true"/>
<Channel className="org.apache.catalina.tribes.group.GroupChannel">
<Membership className="org.apache.catalina.tribes.membership.McastService" address="<228.0.0.4>" port="<45564>" frequency="500" dropTime="3000"/>
<Sender className="org.apache.catalina.tribes.transport.ReplicationTransmitter">
<Transport className="org.apache.catalina.tribes.transport.nio.PooledParallelSender"/>
</Sender>
<Receiver className="org.apache.catalina.tribes.transport.nio.NioReceiver" address="auto" port="4000" autoBind="100" selectorTimeout="5000" maxThreads="6"/>
<Interceptor className="org.apache.catalina.tribes.group.interceptors.TcpFailureDetector"/>
<Interceptor className="org.apache.catalina.tribes.group.interceptors.MessageDispatch15Interceptor"/>
</Channel>
<Valve className="org.apache.catalina.ha.tcp.ReplicationValve" filter=""/>
<Valve className="org.apache.catalina.ha.session.JvmRouteBinderValve"/>
<ClusterListener className="org.apache.catalina.ha.session.JvmRouteSessionIDBinderListener"/>
<ClusterListener className="org.apache.catalina.ha.session.ClusterSessionListener"/>
</Cluster>
Note that we need to add this configuration to each Tomcat instance you wish to add to the cluster as a worker, and that each Engine element must have its own unique jvmRoute.
One possible constraint here is that the tomcat instances talk to each other creating a lot of communications among instances.
To learn more we can refer [http://www.mulesoft.com/tcat/tomcat-clustering]

Persisting Session State to Common Store

We can use common session storage like database, memcached, redis etc to store the session state all the tomcat instances will persist the session state to database or memcached and read session state from there.
  To use memcached as session storage we have to use tomcat version specific session storage manager. Open source contributions around the same are already available here [https://github.com/magro/memcached-session-manager].
 We have to perform the steps below to use memcached as session storage
i.                     put memcached-session-manager-x.y.z.jar, memcached-session-manager-tc<version>-x.y.z.jar, msm-kryo-serializer-x.y.z.jar and spymemcached-x.y.z.jar in ${CATALINA_HOME}/lib directory.
ii.                   edit tomcat/conf/context.xml, and add the following lines inside the <Context> tag (on all nodes):
<Manager className="de.javakaffee.web.msm.MemcachedBackupSessionManager"
      memcachedNodes="n1:localhost:11211"
      requestUriIgnorePattern=".*\.(ico|png|gif|jpg|css|js)$" />
Or we can follow this link to complete the setup [http://wiki.alfresco.com/wiki/Tomcat_Session_Replication_with_Memcached]
For Redis we can follow this link [https://github.com/jcoleman/tomcat-redis-session-manager]


Session Clustering Using Hazelcast

We can use hazelcast to cluster sessions as well. To do so we must have the following -
·         Target application or web server should support Java 1.5+
·         Target application or web server should support Servlet 2.4+ spec.
·         Session objects that need to be clustered have to be Serializable.

Put the hazelcast and hazelcast-wm jars in your WEB-INF/lib directory.
Put the following xml into web.xml file. Make sure Hazelcast filter is placed before all the other filters if any; put it at the top for example
    <filter>
    <filter-name>hazelcast-filter</filter-name>
    <filter-class>com.hazelcast.web.WebFilter</filter-class>
    <init-param>
        <param-name>map-name</param-name>
        <param-value>[my-sessions]</param-value>
    </init-param>
    <init-param>
        <param-name>sticky-session</param-name>
        <param-value>[true]</param-value>
    </init-param>
    <init-param>
        <param-name>debug</param-name>
        <param-value>[true]</param-value>
    </init-param>
</filter>
<filter-mapping>
    <filter-name>hazelcast-filter</filter-name>
    <url-pattern>/*</url-pattern>
    <dispatcher>FORWARD</dispatcher>
    <dispatcher>INCLUDE</dispatcher>
    <dispatcher>REQUEST</dispatcher>
</filter-mapping>
<listener>
    <listener-class>com.hazelcast.web.SessionListener</listener-class>
</listener>
We need to change in hazelcast.xml file also like below –
join>
            <multicast enabled="[false/true]">
                <multicast-group>[224.2.2.3]</multicast-group>
                <multicast-port>[54327]</multicast-port>
            </multicast>
            <tcp-ip enabled="[true/false]">
                <interface>[127.0.0.1]</interface>
<interface>[127.0.0.1]</interface>
            </tcp-ip>
            <aws enabled="[false/true]">
                <access-key>[123]</access-key>
                <secret-key>[456]</secret-key>
                <tag-key>type</tag-key>
                <tag-value>hz-nodes</tag-value>
            </aws>
        </join>


Cross Platform Session Sharing

Some time we while developing enterprise applications we may end up developing a solution involving multiple technologies like Java, J2ee, PHP and Microsoft Technologies for different components and we may expect different components to work seamlessly forcing us to share session for cross platform. To support cross platform session sharing we have to take care of the following things –
i.                     Common session store
ii.                   Common session object structure (data structure)
iii.                 Common session serialization strategy
iv.                 Common deserialization strategy
 We will explore here session sharing strategy between a J2EE based application deployed in tomcat and PHP based application.

Tomcat and J2EE Session Management


Tomcat (J2EE containers) by default has its own session management and storage implementation. It generates a session cookie by the name JSESSIONID and has its own session storage data structure. We need to override this session cookie name as well as data structure and to do so we must have to write some container specific implementations.
Apache Shiro [http://shiro.apache.org/] gives us the facility to override the session management strategy out of the container.
A complete session management strategy including overriding the session cookie, session id generation and persisting the session in memcached and a custom session object (data structure ) can be found here – [https://github.com/badalb/shiro-session-management ]

Session Management from PHP /Zend

Implementing Memcached based session in Zend framework 2 :

/** configuration settings **/
/** File: module.config.php ; Location: <org/com/module/Api/ **/>

defining memcached service:

'service_manager'=>array(
        'factories' => array(
....
....
                   'doctrine.cache.my_memcache' => function ($sm) {
                        $cache = new \Doctrine\Common\Cache\MemcachedCache();                  
                        return $cache;
                    }
.....
)),.....

Defining Session configuration :

....
....
    'session' => array(
        'config' => array(
            'class' => 'Zend\Session\Config\SessionConfig',
            'options' => array(
                'name' => 'hu',
                'cookie_lifetime' => 0,
                'use_cookies' => true,
                'cookie_httponly' => true,
            ),
            'use_memcached'=> true
        ),
        'storage' => 'Zend\Session\Storage\SessionArrayStorage',
        'validators' => array(
            array(
                'Zend\Session\Validator\RemoteAddr',
                'Zend\Session\Validator\HttpUserAgent',
            ),
        ),
    ),
....
....
/** File: module.config.php - END **/

/** File: module.php  ; Location: <org/com/module/Api/config/ **/>

Setting memcached as Storage adapter:
NOTE: Storage adapters come with tons of methods to read, write and modify stored items and to get information about stored items and the storage.

         $cache = StorageFactory::factory(array(
                    'adapter' =>array( 'name' => 'memcached',
                       'options' => array(
                       'servers' => ['localhost'],
                           )
                ),
              ));
        $saveHandler = new \Api\Common\HUCache($cache);
        ini_set('session.serialize_handler', 'php_serialize');

       /**Session config and storage options **/
        $sessionConfig = new SessionConfig();
        $sessionConfig->setOptions($config['session']['config']['options']);

        $sessionManager = new SessionManager($sessionConfig);
        $sessionManager->setStorage(new \Zend\Session\Storage\SessionStorage());

        $sessionManager->setSaveHandler($saveHandler);

/** File: module.php - END **/

/** New File: HUCache.php ; Location : <org/com/module/Api/src/Api/Common/ **/>

/** Storage adapter with read, write and delete methods **/

class HUCache extends \Zend\Session\SaveHandler\Cache
{
    public function read($id)
    {
        $data = Decoder::decode($this->getCacheStorage()->getItem($id), Json::TYPE_ARRAY);
        return serialize($data);
    }

    /**
     * Write session data
     *
     */
    public function write($id, $data)
    {  
        $new_data = (array) unserialize($data);

        $data = Decoder::decode($this->getCacheStorage()->getItem($id));
        $final = $new_data;
        if (null !== $data && !empty($data))
        {
        //Merge the data
            $final = array_merge((array)$data ,$new_data);
        }
        $a = Encoder::encode($final);
        return $this->getCacheStorage()->setItem($id, $a);
    }
        /**
        * Destroy session
        *
        * @param string $id
        * @return bool
        */
    public function destroy($id)
    {
        return $this->getCacheStorage()->removeItem($id);
    }
}
/** File: HUCache.php - END **/