eneter.messaging.messagingsystems.composites.messagebus

Class MessageBusMessagingFactory

java.lang.Object

eneter.messaging.messagingsystems.composites.messagebus.MessageBusMessagingFactory

All Implemented Interfaces:

IMessagingSystemFactory

public class MessageBusMessagingFactory
extends java.lang.Object
implements IMessagingSystemFactory

Extension providing the communication via the message bus. This messaging provides the client-service communication via the message bus. It ensures the communication via the message bus is transparent and for communicating parts it looks like a normal communication via output and input channel.
The duplex input channel created by this messaging will automatically connect the message bus and register the service when the startListening() is called.
The duplex output channel created by this messaging will automatically connect the message bus and ask for the service when the openConnection() is called. The following example shows how to communicate via the message bus.

Implementation of the message bus service that will mediate the client-service communication:

 public class Program
 {
     public static void main(String[] args) throws Exception
     {
         // Message Bus will use TCP for the communication.
         IMessagingSystemFactory aMessaging = new TcpMessagingSystemFactory();
  

         // Input channel to listen to services.
         IDuplexInputChannel aServiceInputChannel = aMessaging.createDuplexInputChannel("tcp://127.0.0.1:8045/");
  

         // Input channel to listen to clients.
         IDuplexInputChannel aClientInputChannel = aMessaging.createDuplexInputChannel("tcp://127.0.0.1:8046/");
  

         // Create the message bus.
         IMessageBus aMessageBus = new MessageBusFactory().createMessageBus();
  

         // Attach channels to the message bus and start listening.
         aMessageBus.attachDuplexInputChannels(aServiceInputChannel, aClientInputChannel);
  

         System.out.println("Message bus service is running. Press ENTER to stop.");
         new BufferedReader(new InputStreamReader(System.in)).readLine();
  

         // Detach channels and stop listening.
         aMessageBus.detachDuplexInputChannels();
     }
 }
 

Implementation of the service which is exposed via the message bus:

 public interface IEcho
 {
     String hello(String text);
 }

 

 ....
 

 // Simple echo service.
 class EchoService implements IEcho
 {
      @Override
     public String hello(String text)
     {
         return text;
     }
  

 }

 

 ....
 

 public class Program
 {
     public static void main(String[] args) throws Exception
     {
         // The service will communicate via Message Bus which is listening via TCP.
         IMessagingSystemFactory aMessageBusUnderlyingMessaging = new TcpMessagingSystemFactory();
         // note: only TCP/IP address which is exposed for services is needed.
         IMessagingSystemFactory aMessaging = new MessageBusMessagingFactory("tcp://127.0.0.1:8045/", null, aMessageBusUnderlyingMessaging);
  

         // Create input channel listening via the message bus.
         // Note: this is address of the service inside the message bus.
         IDuplexInputChannel anInputChannel = aMessaging.createDuplexInputChannel("Eneter.Echo");
  

         // Instantiate class implementing the service.
         IEcho anEcho = new EchoService();
  

         // Create the RPC service.
         IRpcService<IEcho> anEchoService = new RpcFactory().createService(anEcho, IEcho.class);
  

         // Attach input channel to the service and start listening via the message bus.
         anEchoService.attachDuplexInputChannel(anInputChannel);
  

         System.out.println("Echo service is running. Press ENTER to stop.");
         new BufferedReader(new InputStreamReader(System.in)).readLine();
  

         // Detach the input channel and stop listening.
         anEchoService.detachDuplexInputChannel();
     }
 }
 

Implementation of the client using the service which is exposed via the message bus:

 public class Program
 {
     public static void main(String[] args) throws Exception
     {
         // The client will communicate via Message Bus which is listening via TCP.
         IMessagingSystemFactory aMessageBusUnderlyingMessaging = new TcpMessagingSystemFactory();
         // note: only TCP/IP address which is exposed for clients is needed. 
         IMessagingSystemFactory aMessaging = new MessageBusMessagingFactory(null, "tcp://127.0.0.1:8046/", aMessageBusUnderlyingMessaging);
  

         // Create output channel that will connect the service via the message bus..
         // Note: this is address of the service inside the message bus.
         IDuplexOutputChannel anOutputChannel = aMessaging.createDuplexOutputChannel("Eneter.Echo");
  

         // Create the RPC client for the Echo Service.
         IRpcClient<IEcho> aClient = new RpcFactory().createClient(IEcho.class);
  

         // Attach the output channel and be able to communicate with the service via the message bus.
         aClient.attachDuplexOutputChannel(anOutputChannel);
  

         // Get the service proxy and call the echo method.
         IEcho aProxy = aClient.getProxy();
         String aResponse = aProxy.hello("hello");
  

         System.out.println("Echo service returned: " + aResponse);
  

         // Detach the output channel.
         aClient.detachDuplexOutputChannel();
     }
 }
 

Constructor Summary

Constructors

Constructor and Description
MessageBusMessagingFactory(java.lang.String serviceConnctingAddress, java.lang.String clientConnectingAddress, IMessagingSystemFactory underlyingMessaging) Constructs the factory.
MessageBusMessagingFactory(java.lang.String serviceConnctingAddress, java.lang.String clientConnectingAddress, IMessagingSystemFactory serviceUnderlyingMessaging, IMessagingSystemFactory clientUnderlyingMessaging, ISerializer serializer)
MessageBusMessagingFactory(java.lang.String serviceConnctingAddress, java.lang.String clientConnectingAddress, IMessagingSystemFactory underlyingMessaging, ISerializer serializer) Constructs the factory.

Method Summary

Modifier and Type	Method and Description
IDuplexInputChannel	createDuplexInputChannel(java.lang.String channelId) Creates the input channel which can receive and send messages to the output channel.
IDuplexOutputChannel	createDuplexOutputChannel(java.lang.String channelId) Creates the output channel which can sends and receive messages from the input channel.
IDuplexOutputChannel	createDuplexOutputChannel(java.lang.String channelId, java.lang.String responseReceiverId) Creates the output channel which can sends and receive messages from the input channel.
IMessagingSystemFactory	getClientMessaging() Gets messaging used by clients to connect the message bus.
int	getConnectionTimeout() Returns maximum time for opening connection with the service via the message bus.
IThreadDispatcherProvider	getInputChannelThreading() Gets threading mode used for input channels.
IThreadDispatcherProvider	getOutputChannelThreading() Gets threading mode used for output channels.
IMessagingSystemFactory	getServiceMessaging() Gets messaging used by services to be exposed via the message bus.
MessageBusMessagingFactory	setClientMessaging(IMessagingSystemFactory clientMessaging) Sets messaging used by clients to connect the message bus.
MessageBusMessagingFactory	setConnectTimeout(int milliseconds) Sets maximum time for opening connection with the service via the message bus.
MessageBusMessagingFactory	setInputChannelThreading(IThreadDispatcherProvider inputChannelThreading) Sets threading mode for input channels.
MessageBusMessagingFactory	setOutputChannelThreading(IThreadDispatcherProvider outputChannelThreading) Sets threading mode for output channels.
MessageBusMessagingFactory	setServiceMessaging(IMessagingSystemFactory serviceMessaging) messaging used by services to be exposed via the message bus.

Methods inherited from class java.lang.Object

equals, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

MessageBusMessagingFactory

public MessageBusMessagingFactory(java.lang.String serviceConnctingAddress,
                                  java.lang.String clientConnectingAddress,
                                  IMessagingSystemFactory underlyingMessaging)

Constructs the factory.

Parameters:

serviceConnctingAddress - message bus address intended for services which want to register in the message bus. It can be null if the message bus factory is intended to create only duplex output channels.

clientConnectingAddress - message bus address intended for clients which want to connect a registered service. It can be null if the message bus factory is intended to create only duplex input channels.

underlyingMessaging - messaging system used by the message bus.

MessageBusMessagingFactory

public MessageBusMessagingFactory(java.lang.String serviceConnctingAddress,
                                  java.lang.String clientConnectingAddress,
                                  IMessagingSystemFactory underlyingMessaging,
                                  ISerializer serializer)

Constructs the factory.

Parameters:

serviceConnctingAddress - message bus address intended for services which want to register in the message bus. It can be null if the message bus factory is intended to create only duplex output channels.

clientConnectingAddress - message bus address intended for clients which want to connect a registered service. It can be null if the message bus factory is intended to create only duplex input channels.

underlyingMessaging - messaging system used by the message bus.

serializer - serializer which is used to serialize MessageBusMessage which is internally used for the communication with the message bus.

MessageBusMessagingFactory

public MessageBusMessagingFactory(java.lang.String serviceConnctingAddress,
                                  java.lang.String clientConnectingAddress,
                                  IMessagingSystemFactory serviceUnderlyingMessaging,
                                  IMessagingSystemFactory clientUnderlyingMessaging,
                                  ISerializer serializer)

Method Detail

createDuplexOutputChannel

public IDuplexOutputChannel createDuplexOutputChannel(java.lang.String channelId)
                                               throws java.lang.Exception

Description copied from interface: IMessagingSystemFactory

Creates the output channel which can sends and receive messages from the input channel.

Specified by:

createDuplexOutputChannel in interface IMessagingSystemFactory

Parameters:

channelId - address of the input channel.

Returns:

output channel

Throws:

java.lang.Exception

createDuplexOutputChannel

public IDuplexOutputChannel createDuplexOutputChannel(java.lang.String channelId,
                                                      java.lang.String responseReceiverId)
                                               throws java.lang.Exception

Description copied from interface: IMessagingSystemFactory

Creates the output channel which can sends and receive messages from the input channel.

Specified by:

createDuplexOutputChannel in interface IMessagingSystemFactory

Parameters:

channelId - address of the input channel.

responseReceiverId - unique identifier of the output channel. If the value is null then the identifier is genearated automatically

Returns:

duplex output channel

Throws:

java.lang.Exception

createDuplexInputChannel

public IDuplexInputChannel createDuplexInputChannel(java.lang.String channelId)
                                             throws java.lang.Exception

Description copied from interface: IMessagingSystemFactory

Creates the input channel which can receive and send messages to the output channel.

Specified by:

createDuplexInputChannel in interface IMessagingSystemFactory

Parameters:

channelId - address of the input channel.

Returns:

input channel

Throws:

java.lang.Exception

getClientMessaging

public IMessagingSystemFactory getClientMessaging()

Gets messaging used by clients to connect the message bus.

Returns:

setClientMessaging

public MessageBusMessagingFactory setClientMessaging(IMessagingSystemFactory clientMessaging)

Sets messaging used by clients to connect the message bus.

Parameters:

clientMessaging - messaging which shall be used clients to connect the message bus.

Returns:

getServiceMessaging

public IMessagingSystemFactory getServiceMessaging()

Gets messaging used by services to be exposed via the message bus.

Returns:

setServiceMessaging

public MessageBusMessagingFactory setServiceMessaging(IMessagingSystemFactory serviceMessaging)

messaging used by services to be exposed via the message bus.

Parameters:

serviceMessaging - messaging which shall be used by services to expose their API via the message bus.

Returns:

setInputChannelThreading

public MessageBusMessagingFactory setInputChannelThreading(IThreadDispatcherProvider inputChannelThreading)

Sets threading mode for input channels.

Parameters:

inputChannelThreading - threading model

Returns:

this instance of MessageBusMessagingFactory

getInputChannelThreading

public IThreadDispatcherProvider getInputChannelThreading()

Gets threading mode used for input channels.

Returns:

thread dispatcher which is used for input channels.

setOutputChannelThreading

public MessageBusMessagingFactory setOutputChannelThreading(IThreadDispatcherProvider outputChannelThreading)

Sets threading mode for output channels.

Parameters:

outputChannelThreading -

Returns:

this instance of MessageBusMessagingFactory

getOutputChannelThreading

public IThreadDispatcherProvider getOutputChannelThreading()

Gets threading mode used for output channels.

Returns:

thread dispatcher which is used for output channels.

setConnectTimeout

public MessageBusMessagingFactory setConnectTimeout(int milliseconds)

Sets maximum time for opening connection with the service via the message bus. Default value is 30 seconds. When the client opens the connection with a service via the message bus it requests message bus to open the connection with a desired service. The message checks if the requested service exists and if yes it forwards the open connection request. Then when the service receives the open connection request it sends back the confirmation message that the client is connected. This timeout specifies the maximum time which is allowed for sending the open connection request and receiving the confirmation from the service.

Parameters:

milliseconds -

Returns:

this instance of MessageBusMessagingFactory

getConnectionTimeout

public int getConnectionTimeout()

Returns maximum time for opening connection with the service via the message bus. Default value is 30 seconds. When the client opens the connection with a service via the message bus it requests message bus to open the connection with a desired service. The message checks if the requested service exists and if yes it forwards the open connection request. Then when the service receives the open connection request it sends back the confirmation message that the client is connected. This timeout specifies the maximum time which is allowed for sending the open connection request and receiving the confirmation from the service.

Returns:

time in milliseconds