Server Sent Events (SSE) using Jersey, spring and Javascript

educcess Java, Programming, Web Services, Web Technologies Develop Notifications, Implementation of Notifications, java and SSE, javascript and SSE, jersey2.8, Real time notifications in java, Server Sent Events, SSE 3 Comments

Server Side Event not firing in Jersey 2.8 using SSE

Notifications were playing a major role in every applications either it is a mobile application or web application or even a desktop application. These days every latest Operating system updates were including Facebook as a service within OS itself and mail notifications were just on your desktop.

It is important to learn how to support realtime notifications in your web applications. Following technologies were considered in the given example

Springs

Jersey2.8

Javascript

Server Sent Events Javascript Code

Register to Server Sent Events in Javascript

var notificationBaseURL =  "http://myapplication.com/"; //The URL Where your services are hosted
function listenAllEvents() {
	if (typeof (EventSource) !== "undefined") {

		var source = new EventSource(
				notificationBaseURL+"applicationnotifier/sse/events/register/"+loggedInUserName);
		source.onmessage = notifyEvent;
	} else {
		console.log("Sorry no event data sent - ");
	}
}

function notifyEvent(event) {
	var responseJson = JSON.parse(event.data);
	alert("... Notification Received ...");
}

In the above code the URL is specific to user who have logged in. Every user has to register for notification.

Java Spring Code For Server Sent Events(SSE):

//    NotificationHandler.java

package com.applicationnotifier.Notification.WebServices.Impl;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.inject.Singleton;
import javax.ws.rs.Consumes;
import javax.ws.rs.FormParam;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
import javax.ws.rs.core.MediaType;

import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jettison.json.JSONObject;
import org.glassfish.jersey.media.sse.EventOutput;
import org.glassfish.jersey.media.sse.OutboundEvent;
import org.glassfish.jersey.media.sse.SseBroadcaster;
import org.glassfish.jersey.media.sse.SseFeature;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.web.bind.annotation.ResponseBody;

import com.applicationnotifier.Notification.framework.impl.NotificationFrameworkFactory;
import com.applicationnotifier.Notification.framework.intf.NotificationFrameworkInterface;
import com.applicationnotifier.dao.notification.NotificationDao;
import com.applicationnotifier.pojo.Form;
import com.applicationnotifier.pojo.notification.Notification;
import com.applicationnotifier.responsepojo.NotificationResponse;

/*
    A CLASS THAT REGISTERS NOTIFICATIONS
    Registering does following functions
    1. Create a broadcaster object for each notification type
    2. Map Event Output object for each broadcaster to broadcast a message
    3. Broadcast the event
 */

@Singleton
@Path("/events")
public class NotificationHandler {
	
	final static Logger logger = LoggerFactory.getLogger(NotificationHandler.class);
	@Autowired
	@Qualifier("notificationDaoImpl")
	NotificationDao notificationDao;

    /*
        A map thats keeps track of each notification and its output event object. broadcaster object.
        SseBroadcaster will perform broadcasting the notification
     */
	Map<String, SseBroadcaster> notificationBroadcasterMap = new HashMap<String, SseBroadcaster>();

    
    /*
     registerForAnEventSummary: will be called when the client registers for notifications
     in javascript we call listenAllEvents() method.
     */
	@Path("/register/{userName}")
	@Produces(SseFeature.SERVER_SENT_EVENTS)
	@GET
	public @ResponseBody EventOutput registerForAnEventSummary(
			@PathParam("userName") String userName) {
		try {
			NotificationFrameworkFactory factory = new NotificationFrameworkFactory();
			
			EventOutput eventOutput = new EventOutput();
			

            /* Returns all types of notifications, for each type of notification there should be an implementation as newMessageNotificationImplementation or newMessageNotificationFramework
             
                Exmple  newMessageNotification, in this example this notification has a class
                newMessageNotificationFramework.java
             
             */
			List notificationTypes = getAllNotificationTypes();

			for (String notificationType : notificationTypes) {
				NotificationFrameworkInterface notificationInterface = factory
						.getNotifieir(notificationType);
				String keyVal = getKeyVal(notificationType, userName);
				if (!notificationBroadcasterMap.containsKey(keyVal)) {
                    //Add broadcaster to map
					notificationBroadcasterMap.put(keyVal,
							notificationInterface.getBroadcaster());
				}
				
                //Get broadcaster and add event output
				notificationBroadcasterMap.get(keyVal).add(eventOutput);
			}

			return eventOutput;
		} catch (NullPointerException exception) {
			logger.error("Exception Occurred: ", exception);
		}
		return null;
	}

    /*
     getKeyVal : every user must register for each notification
                Notification Key : newMessageNotification_Ram indicates Ram is listening to newMessage notification
     */
	private String getKeyVal(String typeOfEvent, String userName) {

		switch (typeOfEvent) {
		case "newMessageNotification":
        case "likeNotification":
        case "commentNotification":
			return typeOfEvent + "_" + userName;
        		
		default:
			return null;
		}
	}

    //Return different types of notifications supported in your application
	private List getAllNotificationTypes() {
		List notificationTypes = new ArrayList();
		notificationTypes.add("newMessageNotification");
		notificationTypes.add("likeNotification");
		notificationTypes.add("commentNotification");
		return notificationTypes;
	}

    /*
        Just returns a Map object for a given json string
     */
	@SuppressWarnings("unchecked")
	protected HashMap<String, String> getMapFromJson(String message) {
		ObjectMapper mapper = new ObjectMapper();
		HashMap<String, String> value = null;
		try {
			value = mapper.readValue(message, HashMap.class);
		} catch (IOException e) {
			logger.error("Exception Occurred: ", e);
		}
		return value;
	}

    /*
        Broad cast notification to all clients which are registered for notification
     */
	@Path("/broadcast")
	@POST
	@Produces(MediaType.TEXT_PLAIN)
	@Consumes(MediaType.APPLICATION_FORM_URLENCODED)
	public String broadcastNotifications(@FormParam("message") String message) {

		try {
			HashMap<String, String> value = getMapFromJson(message);
			JSONObject responseJson = new JSONObject(value);
			/*
			 * System.out .println("received data: " + message);
			 */
			OutboundEvent.Builder eventBuilder = new OutboundEvent.Builder();
			OutboundEvent event = eventBuilder.name("message")
					.mediaType(MediaType.TEXT_PLAIN_TYPE)
					.data(String.class, responseJson.toString()).build();

            //Things you wish to send to client
			String keyVal = getKeyVal(value.get("ntftyp"),
                                      , value.get("un"));
			System.out.println("broadcasting: " + message + " to: " + keyVal);
			if (notificationBroadcasterMap.get(keyVal) != null) {
				// System.out.println("message is ready for broadcasting");
				notificationBroadcasterMap.get(keyVal).broadcast(event);
			} else
				System.out.println("no broadcaster for: " + keyVal);
		} catch (NullPointerException exception) {
			logger.error("Exception Occurred: ", exception);
		}

		return "Message '" + message + "' has been broadcast.";
	}
}



// NewMessageNotificationBusniessIntf.java

package com.applicationnotifier.Notification.business.intf;

import java.util.HashMap;
import java.util.Map;

import com.applicationnotifier.Notification.framework.impl.NotificationAbstractFramework;

public abstract class NewMessageNotificationBusniessIntf extends NotificationAbstractFramework {
    public boolean notifyNewMessage(HashMap<String, Object> message);
}


// NewMessageNotificationBusniessImpl.java
package com.applicationnotifier.Notification.business.impl;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.Map;
import java.util.ResourceBundle;

import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.NameValuePair;
import org.apache.commons.httpclient.methods.PostMethod;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;

import com.applicationnotifier.Notification.business.intf.NewMessageNotificationBusniessIntf;

/*
    Call these methods when some updates happened in your Database
    Example - Some one sent a message, the messageInsert service will be called (Spring Controller,Service,Repository ) 
    in Service layer create object of NewMessageNotificationBusniessImpl and call these methods to notify
 
    In these methods just call Post/Get methods , these methods are services for your notifications
        URL - sse/events/broadcast/
 
    Any call to sse/events/broadcast/ will call a method defined in NotificationHandler class i.e broadcastNotifications
 */
@Service
public class NewMessageNotificationBusniessImpl extends
		NewMessageNotificationBusniessIntf {
	
    @Override
	public boolean notifyNewMessage(HashMap<String, Object> message) {
		try {
			message.put("msg", message.get("un") + " You have new message ");
			HttpClient httpClient = new HttpClient();

			PostMethod postMethod = null;
            postMethod = new PostMethod(
                    resourceBundle.getString("localhost:8080")
                            + resourceBundle.getString("applicationnotifier")
                            + resourceBundle
                                    .getString("sse/events/broadcast/"));
			
			// postMethod.addParameter(data[0]);
			NameValuePair[] parametersBody = {
					new NameValuePair("message", convertToJson(message)),
					};
			postMethod.setRequestBody(parametersBody);
			httpClient.executeMethod(postMethod);

			BufferedReader responseReader = new BufferedReader(
					new InputStreamReader(postMethod.getResponseBodyAsStream()));
			String line;
			while ((line = responseReader.readLine()) != null) {
				System.out.println(line);
			}
						
			return true;
		} catch (Exception e) {
			logger.error("Exception Occurred: ", e);
		}
		return false;
	}
}


//NotificationAbstractFramework.java
package com.applicationnotifier.Notification.framework.impl;

import java.io.IOException;
import java.util.Map;

import org.codehaus.jackson.map.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public abstract class NotificationAbstractFramework {
	
	final static Logger logger = LoggerFactory.getLogger(NotificationAbstractFramework.class);

	protected String convertToJson(Map<String, Object> message) {
		try {
			ObjectMapper mapper = new ObjectMapper();
			return mapper.writeValueAsString(message);
		} catch (IOException e) {
			logger.error("Exception Occurred: ", e);
		}
		return "";
	}
}