Flex data push with BlazeDS and EJB3

With BlazeDS we can push data from the J2EE container to the Flex clients just like Adobe Life Cycle Data Services can. In blog I will use the Message Broker and Service Adapter of BlazeDS in combination with EJB3 Entity / Session Bean to push every change to the client. The Message Broker routes the messages with the EJB entities to the custom service adapter. The service Adapter is a service, which registers all the subscribed Flex clients and publish the changes to these client.

For this example I used JDeveloper 11g and Weblogic. ( For more information see my previous blog ) .
This are the steps we need to do to make this work.

• Generate an Entity Bean with Entity from Tables in JDeveloper
  
• Generate an EJB Session Bean and Add the Message Broker code
  
• Make an ServiceAdapter
• Configure the BlazeDS files
• Add an Flex Class which maps to the Entity Bean
• Add the Flex producer and consumer code
• Create an EJB client for the changes.
  

We start by creating an Employee Entity Bean. ( Based on the employees table of the Oracle HR schema )

package nl.whitehorses.model.entities;

import java.io.Serializable;
import java.sql.Timestamp;

import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.Id;
import javax.persistence.NamedQueries;
import javax.persistence.NamedQuery;


@Entity
@NamedQueries( { @NamedQuery(name = "Employees.findAll", query = "select o from Employees o") })
public class Employees implements Serializable {
@Column(name="COMMISSION_PCT")
private Double commissionPct;
@Column(name="DEPARTMENT_ID")
private Long departmentId;
@Column(nullable = false, unique = true, length = 25)
private String email;
@Id
@Column(name="EMPLOYEE_ID", nullable = false)
private Long employeeId;
@Column(name="FIRST_NAME", length = 20)
private String firstName;
@Column(name="HIRE_DATE", nullable = false)
private Timestamp hireDate;
@Column(name="JOB_ID", nullable = false, length = 10)
private String jobId;
@Column(name="LAST_NAME", nullable = false, length = 25)
private String lastName;
@Column(name="PHONE_NUMBER", length = 20)
private String phoneNumber;
private Double salary;


public Employees() {
}

  .......
}

My HRSessionEJBBean Session Bean, this Bean has a remote interface. JDeveloper can generate this Session Bean for you with its remote interface. This Bean contains the Message Broker code which routes the messages with all the employees when the persist, merge or remove method of the EntityManager is called.

package nl.whitehorses.model.services;

import java.util.List;

import javax.ejb.Remote;
import javax.ejb.Stateless;

import javax.persistence.EntityManager;
import javax.persistence.PersistenceContext;

import nl.whitehorses.model.entities.Employees;

import flex.messaging.MessageBroker;
import flex.messaging.messages.AsyncMessage;
import flex.messaging.util.UUIDUtils;

@Stateless(name = "HRSessionEJB", mappedName = "flex_ejb2-Model-HRSessionEJB")
@Remote
public class HRSessionEJBBean implements HRSessionEJB {
 @PersistenceContext(unitName="Model")
 private EntityManager em;

 public HRSessionEJBBean() {
 }

 private void pushEmployees() {
     String clientId = UUIDUtils.createUUID();
     MessageBroker msgBroker = MessageBroker.getMessageBroker(null);
     AsyncMessage msg = new AsyncMessage();
     msg.setDestination("EmployeesServicePush");
     msg.setClientId(clientId);
     msg.setMessageId(UUIDUtils.createUUID());
     msg.setBody(getEmployeesFindAll());
     msgBroker.routeMessageToService(msg,null);     
 }

 public Employees persistEmployees(Employees employees) {
     em.persist(employees);
     pushEmployees();
     return employees;
 }

 public Employees mergeEmployees(Employees employees) {
    Employees emp = em.merge(employees);
    pushEmployees();
    return emp;
 }

 public void removeEmployees(Employees employees) {
     employees = em.find(Employees.class, employees.getEmployeeId());
     em.remove(employees);
     pushEmployees();
 }

 /** <code>select o from Employees o</code> */
 public List<Employees> getEmployeesFindAll() {
     System.out.println("getEmps");
     return em.createNamedQuery("Employees.findAll").getResultList();
 }
}

Create the Blaze Custom Service Adapter, we will add this adapter later to the BlazeDS configuration files. This adapter will do a JNDI lookup of our EJB Session Bean and pushes the messages to the connected Flex clients.

package nl.whitehorses.blazeds.adapter;

import java.util.List;

import flex.messaging.messages.AsyncMessage;
import flex.messaging.messages.Message;
import flex.messaging.services.MessageService;
import flex.messaging.services.ServiceAdapter;


import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

import nl.whitehorses.model.entities.Employees;
import nl.whitehorses.model.services.HRSessionEJB;

public class EmployeesServiceAdapter extends ServiceAdapter {
 Context context = null;
 HRSessionEJB hRSessionEJB = null;

 public EmployeesServiceAdapter() {
     try {
         context = new InitialContext();
         hRSessionEJB =
                 (HRSessionEJB)context.lookup("flex_ejb2-Model-HRSessionEJB#nl.whitehorses.model.services.HRSessionEJB");
     } catch (NamingException e) {
         e.printStackTrace();
     }
     System.out.println("Adapter initilized");
 }

 public void start() {
     System.out.println("Adapter started");

 }

 public void stop() {
     System.out.println("Adapter stopped");
 }

 private List<Employees> getEmployees() {
     return hRSessionEJB.getEmployeesFindAll();
 }

 public Object invoke(Message msg) {
     if (msg.getBody().equals("New")) {
         System.out.println("Adapter received new");
         return getEmployees();
     } else {
         System.out.println("Adapter sending message");
         AsyncMessage newMessage = (AsyncMessage)msg;
         MessageService msgService =
             (MessageService)getDestination().getService();
         msgService.pushMessageToClients(newMessage, true);
     }
     return null;
 }
}

In the messaging-config.xml we need add this custom Service Adapter and add a Employee destination which connect this to our custom adapter and the streaming AMF channel

<?xml version="1.0" encoding="UTF-8"?>
<service id="message-service" class="flex.messaging.services.MessageService">

<adapters>
    <adapter-definition id="EmployeesServicePushAdapter" class="nl.whitehorses.blazeds.adapter.EmployeesServiceAdapter"/>
</adapters>

<destination id="EmployeesServicePush">
  <channels>
    <channel ref="my-streaming-amf" />
  </channels>
  <adapter ref="EmployeesServicePushAdapter"/>
</destination>

</service>

The services-config.xml which import the messaging-config.xml configuration file and contains the streaming-amf channel configuration.

<?xml version="1.0" encoding="UTF-8"?>
<services-config>
<services>
    <service-include file-path="messaging-config.xml" />
    <default-channels>
        <channel ref="my-amf"/>
    </default-channels>
</services>

<channels>
    <channel-definition id="my-amf"
                        class="mx.messaging.channels.AMFChannel">
        <endpoint url="http://{server.name}:{server.port}/{context.root}/messagebroker/amf"
                  class="flex.messaging.endpoints.AMFEndpoint"/>
        <properties>
            <polling-enabled>false</polling-enabled>
        </properties>
    </channel-definition>

    <channel-definition id="my-streaming-amf"
                        class="mx.messaging.channels.StreamingAMFChannel">
        <endpoint url="http://{server.name}:{server.port}/{context.root}/messagebroker/streamingamf"
                  class="flex.messaging.endpoints.StreamingAMFEndpoint"/>
        <properties>
            <idle-timeout-minutes>0</idle-timeout-minutes>
            <max-streaming-clients>10</max-streaming-clients>
            <server-to-client-heartbeat-millis>5000</server-to-client-heartbeat-millis>
            <user-agent-settings>
               <user-agent match-on="MSIE" kickstart-bytes="2048" max-streaming-connections-per-session="3"/>
               <user-agent match-on="Firefox" kickstart-bytes="2048" max-streaming-connections-per-session="3"/>
            </user-agent-settings>
        </properties>
    </channel-definition>
</channels>

<logging>
    <target class="flex.messaging.log.ConsoleTarget" level="Error">
        <properties>
            <prefix>[Flex]</prefix>
            <includeDate>false</includeDate>
            <includeTime>false</includeTime>
            <includeLevel>false</includeLevel>
            <includeCategory>false</includeCategory>
        </properties>
    </target>
</logging>
</services-config>

We are finished with the java part and we can work on the Flex part.
In Flex we need to add a simple Employees class which is connected to the entity bean

package entities
{
 [Bindable]
 [RemoteClass(alias="nl.whitehorses.model.entities.Employees")]
 public class Employees
 {
     public var commissionPct:Number;
     public var departmentId:int;
     public var email:String;
     public var employeeId:int;
     public var firstName:String;
     public var hireDate:Date;
     public var jobId:int;
     public var lastName:String;
     public var phoneNumber:String;
     public var salary:Number;


     public function Employees()
     {
     }

 }
}

The Flex application mxml with the producer and consumer component

<?xml version="1.0" encoding="utf-8"?>
<mx:Application xmlns:mx="http://www.adobe.com/2006/mxml" width="1200" height="500">
<mx:Script>
 <![CDATA[
  import mx.collections.ArrayCollection;
   import mx.messaging.messages.IMessage;
   import mx.messaging.events.MessageAckEvent;
   import mx.messaging.messages.AsyncMessage;
   import mx.messaging.events.MessageEvent;
   import entities.Employees;

   private function init():void{
    var message:AsyncMessage = new AsyncMessage();
          message.body = "New";
          producer.send(message);
    consumer.subscribe();
   }

   private function onMsg(event:MessageEvent):void{
    grid.dataProvider = event.message.body as ArrayCollection;
   }

   private function pub():void   {
          var message:AsyncMessage = new AsyncMessage();
          message.body = "New";
          producer.send(message);
   }

   private function ack(event:MessageAckEvent):void{
    grid.dataProvider = event.message.body as ArrayCollection;
   }

 ]]>
</mx:Script>
<mx:applicationComplete>init();</mx:applicationComplete>

 <mx:Producer id="producer" destination="EmployeesServicePush" acknowledge="ack(event)"/>
 <mx:Consumer id="consumer" destination="EmployeesServicePush" message="onMsg(event)"/>

 <mx:VBox>
  <mx:DataGrid id="grid">
   <mx:columns>
    <mx:DataGridColumn dataField="firstName" headerText="First Name"/>
    <mx:DataGridColumn dataField="lastName" headerText="Last Name"/>
    <mx:DataGridColumn dataField="departmentId" headerText="Department"/>
   </mx:columns>
  </mx:DataGrid>
 </mx:VBox> 


 
</mx:Application>

We can start the J2EE container and the Flex client. To test this we can use an EJB test client which adds a new Employee and look in the Flex application if we can see this new Employee.

package test;

import java.util.Calendar;
import java.util.Hashtable;

import javax.naming.Context;
import javax.naming.InitialContext;

import javax.naming.NamingException;

import nl.whitehorses.model.entities.Employees;
import nl.whitehorses.model.services.HRSessionEJB;

public class HRSessionEJBClient {
public static void main(String [] args) {
    try {
        final Context context = getInitialContext();
        HRSessionEJB hRSessionEJB = (HRSessionEJB)context.lookup("flex_ejb2-Model-HRSessionEJB#nl.whitehorses.model.services.HRSessionEJB");

        Employees emp = new Employees();
        emp.setEmployeeId(995L);
        emp.setDepartmentId(50L);
        emp.setFirstName("a");
        emp.setLastName("a");
        emp.setJobId("SH_CLERK");
        emp.setEmail("a");
        emp.setHireDate(new java.sql.Timestamp(Calendar.getInstance().getTime().getTime()));
        hRSessionEJB.mergeEmployees(emp);

    } catch (Exception ex) {
        ex.printStackTrace();
    }
}

private static Context getInitialContext() throws NamingException {
    Hashtable env = new Hashtable();
    // WebLogic Server 10.x connection details
    env.put( Context.INITIAL_CONTEXT_FACTORY, "weblogic.jndi.WLInitialContextFactory" );
    env.put(Context.PROVIDER_URL, "t3://127.0.0.1:7101");
    return new InitialContext( env );
}
}