Android: How to Receive Notification Messages from Multiple .NET Applications via Websockets

Summary: Simple example showing how an Android application can subscribe to receive notification messages from multiple .NET applications using Websockets.

The example demonstrated in the article can be downloaded here.

Introduction

This is a free continuation of the article Android: How to communicate with .NET application via TCP that describes the request-response communication between Android and .NET application.

The following article demonstrates a different type of scenario. It shows the publish-subscribe communication where Android registers to receive notification messages that can come from multiple .NET applications.
It means the Android application is a subscriber asynchronously receiving specified notifications from publishing .NET applications.

The main point in this scenario is to keep subscribers and publishers decoupled so that they do not have assumptions about each other. That subscribing application does not have to know about applications sending notification messages and also publishing applications do not have to know about receivers of their notifications.
To keep them decoupled the example implements the broker service that will be able to register subscribers and forward them messages from publishers. Subscribers will then use this broker to subscribe for whatever messages they want and publishers will send their notification messages only to this broker not knowing to whom they will be forwarded.

The communication in the example is based on the Websocket protocol that provides true full-duplex communication (without polling or long polling overhead) in environments with firewalls allowing only Internet HTTP connections.
(Websocket starts as an ordinary HTTP request but then the TCP connection stays open what allows pushing messages to the client.)


The example bellow uses Eneter Messaging Framework providing functionality for cross-platform interprocess communication capable to connect Android applications with .NET applications. And it also supports communication using Websockets.

(The framework is free for non-commercial use and can be downloaded from http://www.eneter.net/ProductDownload.htm. You need to download Eneter for .NET and Eneter for Android.
More technical details can be found at technical info.)

Android Specifics

Implementing the publish-subscribe scenario for the Android device is much more complicated than for a standalone personal computer. The Android implementation must consider following specifics:

• Android can restart the application when the configuration is changed e.g. when the screen orientation is changed.
• Android can switch to the sleeping mode and CPU is turned off.
• Depending on the device settings WiFi can turn off after several minutes of inactivity (e.g. during the sleeping mode).

Android can restart the application when the configuration is changed

The life-cycle of an Android application is not so simple as the lifecycle of a standalone desktop application. The life of the Android application goes through several states and the application can be restarted during its life.
The typical example is when the Android device changes its orientation e.g. from portrait to landscape. The application is restarted in order to provide the layout for the landscape position. It is then responsibility of the application to handle this situation and recover from the restart.

In our scenario we need to ensure that during restarting (e.g. when user changed the device orientation) the network connection is not lost and the application will continue receiving subscribed messages. To ensure this behavior, we will use functionality called Fragments. The Fragment is a class whose instance can be set to be NOT destroyed during the restart. Therefore, we will implement a non-UI Fragment where will be located logic for the network communication. After the restart the application activity can obtain the Fragment that holds the network connection and so continue in processing of notification messages.

The small inconvenience with Fragments is this functionality is available from Android 3.0 (API Level 11) and so it is not yet available for the most devices on the market.
Fortunately there is available a library exposing this functionality from Android 1.6. You can install it using 'Android SDK Manager' (see the picture bellow) and then include android-support-v13.jar library directly into your project.

Android can switch to the sleeping mode and CPU is turned off

In order to save batteries the Android device may switch to the sleeping mode turning the CPU off. In this state the application is not running and so it cannot process incoming notification messages. To resolve this situation the following alternatives could be considered:

1. Ignoring the sleeping mode. Although the CPU is off the WiFi antenna is still on (depends on settings, it can stay on several minutes or forever) and incoming messages are received and buffered. Then later on when the device is woken up, notification messages are pushed to the application to be processed.
2. Using AlarmManager to wake CPU up on a regular basis and invoke the code to perform some activities. However, in our case the Android application is a subscriber waiting for notification messages to be pushed from the broker and there are no explicit activities we could schedule to receive them (we do not do polling in our scenario). So using AlarmManager is probably not useful for our communication.
3. Using PowerManager with PARTIAL_WAKE_LOCK to keep the CPU on. This alternative would work but you should keep in mind this would have a significant impact on the battery consumption.
4. Using GCM (former C2DM). It is the Google service supported by the Android platform. It provides the possibility to send a short notification messages to the android application. Typically this notification message is intended to inform the Android application about available data on the server. Android application is then responsible for connecting its server and requesting data. Nice is that in case the notification message is received in the sleeping mode the CPU can be woken up and the application gets the control to handle the incoming notification. The drawback for this alternative could be:
• Performance - it can cross four process boundaries to get data to Android:
  Service --notify--> GCM --notify--> Android --request data--> Service --response data--> Android.
• Dependency on a foreign service - using GCM makes software dependent on the Google service.

For the example bellow I chose the first alternative. It means if the CPU is sleeping the messages are received but will be processed later on when the CPU is woken up.
(To be honest, originally I wanted to achieve CPU is woken up when WiFi receives some data. E.g. if the device has a TCP connection via WiFi and goes to sleep, then when a TCP packet is received the CPU is woken up so that the android application can process received data. Unfortunately, I did not find a way how to implement it and I am not sure if it is possible. I would be very interested if you have some ideas here.)

WiFi can be turned off after several minutes of inactivity

Depending on the WiFi sleeping policy settings, WiFi can be turned off after several minutes of inactivity (it can be also set to never go to sleep). If the WiFi antenna is turned off the connection is broken and then later on when the CPU is woken up the application must create the new connection with the broker and also it must subscribe again to receive notification messages.

The example bellow shows how to be informed about the changes in the connection and displays the status on the screen. It also shows how to recreate the connection with the broker.

Other Android Specifics

1. Do not forget to set INTERNET permssion for your Android application.
2. Use the special IP address 10.0.2.2 if the Android emulator needs to access the service running on local computer and exposed via 127.0.0.1 (loopback).

For more details please refer to my previous article Android: How to communicate with .NET application via TCP.

Broker Service

The broker is a simple .NET console application (service) receiving messages from publishing applications and forwarding them to all subscribed receivers. The subscribed receiver in our scenario is the Android application.

using System;
using Eneter.Messaging.MessagingSystems.MessagingSystemBase;
using Eneter.Messaging.MessagingSystems.WebSocketMessagingSystem;
using Eneter.Messaging.Nodes.Broker;

namespace BrokerApplication
{
    class Program
    {
        static void Main(string[] args)
        {
            // Create the broker.
            IDuplexBrokerFactory aBrokerFactory = new DuplexBrokerFactory();
            IDuplexBroker aBroker = aBrokerFactory.CreateBroker();

            // Create messaging based on Websockets.
            IMessagingSystemFactory aMessaging = new WebSocketMessagingSystemFactory();
            IDuplexInputChannel aBrokerInputChannel = 
              aMessaging.CreateDuplexInputChannel("ws://127.0.0.1:8095/MyBroker/");

            // Attach the input channel to the broker and start listening.
            aBroker.AttachDuplexInputChannel(aBrokerInputChannel);

            // Clients can use the broker now to publish and subscribe messages.
            // Note: When the broker receives a message it forwards it to clients that are
            //       subscribed for this type of message.
            //       Clients that are not subscribed to that type of message will not receive it.
            Console.WriteLine("The broker is running. Press ENTER to stop.");
            Console.ReadLine();

            // Detach the input channel and stop listening.
            aBroker.DetachDuplexInputChannel();
        }
    }
}

Publisher - .NET Application

The publisher is a .NET application providing simple UI to send three types of notification messages to the broker. The broker receives messages and forwards them to subscribed clients.

using System;
using System.Windows.Forms;
using Eneter.Messaging.DataProcessing.Serializing;
using Eneter.Messaging.MessagingSystems.MessagingSystemBase;
using Eneter.Messaging.MessagingSystems.WebSocketMessagingSystem;
using Eneter.Messaging.Nodes.Broker;

namespace Publisher
{
    public partial class Form1 : Form
    {
        // Notification message 1
        public class NotifyMsg1
        {
            public string CurrentTime { get; set; }
        }

        // Notification message 2
        public class NotifyMsg2
        {
            public int Number { get; set; }
        }

        // Notification message 3
        public class NotifyMsg3
        {
            public string TextMessage { get; set; }
        }


        // Broker client is used to send messages to the broker,
        // that forwards messages to subscribers.
        private IDuplexBrokerClient myBrokerClient;

        // Connects broker client with the broker.
        private IDuplexOutputChannel myOutputChannel;

        // Serializer used to serialize notification messages.
        private XmlStringSerializer mySerializer = new XmlStringSerializer();


        public Form1()
        {
            InitializeComponent();

            // Create broker client responsible for sending messages to the broker.
            IDuplexBrokerFactory aBrokerFactory = new DuplexBrokerFactory();
            myBrokerClient = aBrokerFactory.CreateBrokerClient();

            // Create output channel to send messages via websockets.
            IMessagingSystemFactory aMessaging = new WebSocketMessagingSystemFactory();
            myOutputChannel = 
              aMessaging.CreateDuplexOutputChannel("ws://127.0.0.1:8095/MyBroker/");

            // Register to be informed about connection status with the broker.
            myOutputChannel.ConnectionOpened += OnConnectionOpened;
            myOutputChannel.ConnectionClosed += OnConnectionClosed;

            try
            {
                // Attach the output channel to the broker client to be able to send messages.
                myBrokerClient.AttachDuplexOutputChannel(myOutputChannel);
            }
            catch
            {
                // if opening connection failed then it can be reopen with
                // the 'Reconnect' button.
            }
        }

        // Correctly close the output channel.
        private void Form1_FormClosed(object sender, FormClosedEventArgs e)
        {
            // Detach the input channel and stop the listening.
            // Note: It releases the listening thread.
            myBrokerClient.DetachDuplexOutputChannel();
        }

        // Indicates the connection with the broker was open.
        private void OnConnectionOpened(object sender, DuplexChannelEventArgs e)
        {
            DisplayConnectionStatus(true);
        }

        // Indicates the connection with the broker was closed.
        private void OnConnectionClosed(object sender, DuplexChannelEventArgs e)
        {
            DisplayConnectionStatus(false);
        }

        private void DisplayConnectionStatus(bool isConnected)
        {
            string aStatus = isConnected ? "Connected" : "Disconnected";

            // Enforce manipulating UI from the UI thread.
            UI(() => StatusLabel.Text = aStatus);
        }

        // Send NotifyMsg1
        private void Notify1Btn_Click(object sender, EventArgs e)
        {
            NotifyMsg1 aMsg = new NotifyMsg1();
            aMsg.CurrentTime = DateTime.Now.ToString();

            object aSerializedMsg = mySerializer.Serialize<NotifyMsg1>(aMsg);

            myBrokerClient.SendMessage("MyNotifyMsg1", aSerializedMsg);
        }

        // Send NotifyMsg2
        private void Notify2Btn_Click(object sender, EventArgs e)
        {
            NotifyMsg2 aMsg = new NotifyMsg2();
            aMsg.Number = 12345;

            object aSerializedMsg = mySerializer.Serialize<NotifyMsg2>(aMsg);

            myBrokerClient.SendMessage("MyNotifyMsg2", aSerializedMsg);
        }

        // Send NotifyMsg3
        private void Notify3Btn_Click(object sender, EventArgs e)
        {
            NotifyMsg3 aMsg = new NotifyMsg3();
            aMsg.TextMessage = "My notifying text message.";

            object aSerializedMsg = mySerializer.Serialize<NotifyMsg3>(aMsg);

            myBrokerClient.SendMessage("MyNotifyMsg3", aSerializedMsg);
        }

        // Reestablishes the connection with the broker.
        private void ReconnectBtn_Click(object sender, EventArgs e)
        {
            myBrokerClient.DetachDuplexOutputChannel();
           
            try
            {
                // Attach the output channel to the broker client to be able to send messages.
                myBrokerClient.AttachDuplexOutputChannel(myOutputChannel);
            }
            catch
            {
                // if opening connection failed then it can be reopen with
                // the 'Reconnect' button.
            }
        }

        // Helper method to invoke some functionality in UI thread.
        private void UI(Action uiMethod)
        {
            // If we are not in the UI thread then we must synchronize via the invoke mechanism.
            if (InvokeRequired)
            {
                Invoke(uiMethod);
            }
            else
            {
                uiMethod();
            }
        }

    }
}

Subscriber - Android Application

The Android application implements the functionality receiving notification messages that are pushed from the broker service. It provides a simple UI allowing to subscribe for specific messages. The functionality responsible for the communication across the network is implemented in the Fragment ensuring so the network connection is not broken if the application is restarted (e.g. if the device orientation is changed from portrait to landscape). If the device goes to the sleeping mode messages are received (until WiFi is not sleeping) but not processed. Then later on when the device is woken up the received messages are processed by the application. If WiFi sleeps too, the connection is broken what is then detected by the application when woken up. User has then the possibility to recover the connection with the broker.

package eneter.android;

import java.util.*;

import android.os.Bundle;
import android.support.v4.app.Fragment;
import eneter.messaging.diagnostic.EneterTrace;
import eneter.messaging.messagingsystems.messagingsystembase.*;
import eneter.messaging.messagingsystems.websocketmessagingsystem.WebSocketMessagingSystemFactory;
import eneter.messaging.nodes.broker.*;
import eneter.net.system.*;

// Implements the communication with the broker as a non-UI fragment.
// It ensures the communication will not be interrupted e.g. in case
// the device changed the screen orientation.
public class BrokerClientFragment extends Fragment
{
    // Eneter communication.
    private IDuplexBrokerClient myBrokerClient;
    private IDuplexOutputChannel myOutputChannel;
    
    // Notification messages subscribed in the broker.
    private Set<String> mySubscribedNotifications = Collections.synchronizedSet(new HashSet<String>());

    // Events that will be pushed to the activity.
    private EventImpl<Boolean> myConnectionStatusChangedEvent = new EventImpl<Boolean>();
    private EventImpl<BrokerMessageReceivedEventArgs> myBrokerMessageReceived =
            new EventImpl<BrokerMessageReceivedEventArgs>();
    
    
    public BrokerClientFragment() throws Exception
    {
        //EneterTrace.setDetailLevel(EDetailLevel.Debug);
        
        // Instantiate the broker client.
        IDuplexBrokerFactory aBrokerFactory = new DuplexBrokerFactory();
        myBrokerClient = aBrokerFactory.createBrokerClient();
        
        // Handle notification messages from the bropker.
        myBrokerClient.brokerMessageReceived().subscribe(myOnBrokerMessageReceived);
        
        // Let's use Websocket messaging.
        // Note: 10.0.2.2 is a special alias to access the loopback (127.0.0.1)
        //       on the development machine from the emulator.
        IMessagingSystemFactory aMessaging = new WebSocketMessagingSystemFactory();
        myOutputChannel = 
                aMessaging.createDuplexOutputChannel("ws://10.0.2.2:8095/MyBroker/");

        // Observe the connection status.
        myOutputChannel.connectionClosed().subscribe(myOnConnectionClosed);
        myOutputChannel.connectionOpened().subscribe(myOnConnectionOpened);
    }

    // Activity registers here to observe the connection
    // with the broker.
    public Event<Boolean> connectionStatusChanged()
    {
        return myConnectionStatusChangedEvent.getApi();
    }
    
    // Activity registers here to process notification messages
    // received from the broker.
    public Event<BrokerMessageReceivedEventArgs> notifyMessageReceived()
    {
        return myBrokerMessageReceived.getApi();
    }
    
    // Called when the Activity.onCreate() has returned.
    // It opens the connection with the broker.
    @Override
    public void onCreate(Bundle savedInstanceState)
    {
        super.onCreate(savedInstanceState);
        
        // Set to keep this fragment during the restart.
        setRetainInstance(true); 
        
        // Open connection with the broker.
        try
        {
            // Attach the output channel and be able to communicate with broker.
            myBrokerClient.attachDuplexOutputChannel(myOutputChannel);
        }
        catch (Exception err)
        {
            EneterTrace.warning("Opening the connection failed.", err);
        }
    }
    
    // Called when the Activity is destroyed.
    // Note: setRetainInstance(true) was applied, so it is not called
    //       when the Activity is just restarted. 
    //       (e.g. if the device changed the orientation)
    @Override
    public void onDestroy()
    {
        // Close websocket connection.
        // Stop listening to response messages.
        myBrokerClient.detachDuplexOutputChannel();
        
        super.onDestroy();
    }
    
    // Subscribes in the broker for specified message type.
    public void subscribeInBroker(String eventType) throws Exception
    {
        // Subscribe for events in the broker.
        myBrokerClient.subscribe(eventType);
        
        // Store the subscribed event type.
        mySubscribedNotifications.add(eventType);
    }
    
    // Unsubscribes from the broker the specified event type.
    public void unsubscribeFromBroker(String eventType) throws Exception
    {
        // Unsubscribe the given event from the broker.
        myBrokerClient.unsubscribe(eventType);
        
        // Remove the event from the storage.
        mySubscribedNotifications.remove(eventType);
    }
    
    // Unsubscribes from the broker all events.
    public void unsubscribe() throws Exception
    {
        // Unsubscribe all events from the broker.
        myBrokerClient.unsubscribe();
        
        // Clear the storage.
        mySubscribedNotifications.clear();
    }
    

    // Recovers the connection with the broker.
    public void recoverConnection()
    {
        try
        {
            myBrokerClient.detachDuplexOutputChannel();
            myBrokerClient.attachDuplexOutputChannel(myOutputChannel);
            
            // if messages were subscribed before the connection
            // got broken then recover the subscription too.
            if (!mySubscribedNotifications.isEmpty())
            {
                String[] aSubscribedEvents = mySubscribedNotifications.toArray(new String[0]);
                myBrokerClient.subscribe(aSubscribedEvents);
            }
        }
        catch (Exception err)
        {
            EneterTrace.warning("Recovering the connection failed.", err);
        }
    }
    
    // Returns true if the connection with the broker is established.
    public boolean isConnected()
    {
        return myBrokerClient.isDuplexOutputChannelAttached() &&
                myBrokerClient.getAttachedDuplexOutputChannel().isConnected();
    }
    
 
    private void onConnectionChanged(boolean isConnectionOpen)
    {
        try
        {
            // Push information about broken connection to the subscribed activity.
            myConnectionStatusChangedEvent.raise(this, isConnectionOpen);
        }
        catch (Exception err)
        {
            EneterTrace.error("ConnectionChanged handler detected error.", err);
        }
    }
    
    
    // Processes when open connection is indicated.
    private EventHandler<DuplexChannelEventArgs> myOnConnectionOpened =
            new EventHandler<DuplexChannelEventArgs>()
            {
                @Override
                public void onEvent(Object sender, DuplexChannelEventArgs e)
                {
                    onConnectionChanged(true);
                }
            };
    
    // Processes when close connection is indicated.
    private EventHandler<DuplexChannelEventArgs> myOnConnectionClosed =
            new EventHandler<DuplexChannelEventArgs>()
            {
                @Override
                public void onEvent(Object sender, DuplexChannelEventArgs e)
                {
                    onConnectionChanged(false);
                }
            };
            
    // Processes when a notfication message from the broker is received.
    private EventHandler<BrokerMessageReceivedEventArgs> myOnBrokerMessageReceived =
            new EventHandler<BrokerMessageReceivedEventArgs>()
            {
                @Override
                public void onEvent(Object sender, BrokerMessageReceivedEventArgs e)
                {
                    try
                    {
                        // Push the message to the subscribed activity.
                        myBrokerMessageReceived.raise(this, e);
                    }
                    catch (Exception err)
                    {
                        EneterTrace.error("MessageReceived handler detected error.", err);
                    }
                }
            };
}

package eneter.android;

import eneter.messaging.dataprocessing.serializing.*;
import eneter.messaging.diagnostic.EneterTrace;
import eneter.messaging.nodes.broker.*;
import eneter.net.system.*;
import android.os.*;
import android.support.v4.app.*;
import android.view.View;
import android.view.View.OnClickListener;
import android.widget.*;

// Note: The example is based on API Level 7.
//       So if we want to use the Fragment functionality, we need
//       to use 'Android Compatibility Package' and its library
//       android-support-v13.jar.
//       Therefore the Activity is derived from FragmentActivity
//       instead of Activity.
public class AndroidConsumerActivity extends FragmentActivity
{
    // Notification message 1
    public static class NotifyMsg1
    {
        public String CurrentTime;
    }

    // Notification message 2
    public static class NotifyMsg2
    {
        public int Number;
    }

    // Notification message 3
    public static class NotifyMsg3
    {
        public String TextMessage;
    }
    
    
    // For marshaling into the UI thread.
    private Handler myRefresh = new Handler();
    
    // UI control displaying incoming messages.
    private EditText myMessage1EditText;
    private EditText myMessage2EditText;
    private EditText myMessage3EditText;
    private TextView myConnectionStatusTextView;
    
    // Fragment surviving restarts.
    private BrokerClientFragment myBrokerClientFragment;
    
    // For deserializing incoming messages.
    private ISerializer mySerializer = new XmlStringSerializer();
    
    
    // Called when the Activiry is first time created or restarted.
    // (e.g. when the device changed the screen orientation) 
    @Override
    public void onCreate(Bundle savedInstanceState)
    {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.main);
        
        // Get UI controls.
        myMessage1EditText = (EditText) findViewById(R.id.received1EditText);
        myMessage2EditText = (EditText) findViewById(R.id.received2EditText);
        myMessage3EditText = (EditText) findViewById(R.id.received3EditText);
        myConnectionStatusTextView = (TextView) findViewById(R.id.statusTextView);
        Button aSubscribe1Btn = (Button) findViewById(R.id.subscribe1Btn);
        Button anUnsubscribe1Btn = (Button) findViewById(R.id.unsubscribe1Btn);
        Button aSubscribe2Btn = (Button) findViewById(R.id.subscribe2Btn);
        Button anUnsubscribe2Btn = (Button) findViewById(R.id.unsubscribe2Btn);
        Button aSubscribe3Btn = (Button) findViewById(R.id.subscribe3Btn);
        Button anUnsubscribe3Btn = (Button) findViewById(R.id.unsubscribe3Btn);
        Button aReconnectBtn = (Button) findViewById(R.id.reconnectBtn);
        
        try
        {
            // Try to get the broker fragment.
            FragmentManager aFragmentManager = getSupportFragmentManager();
            myBrokerClientFragment = 
                    (BrokerClientFragment) aFragmentManager.findFragmentByTag("MyBroker");
            
            // If the broker fragment is not there then this is the
            // first time start.
            if (myBrokerClientFragment == null)
            {
                // Create and register the broker fragment that will survive
                // restarting this activity.
                myBrokerClientFragment = new BrokerClientFragment();
                
                FragmentTransaction aTransaction = aFragmentManager.beginTransaction();
                aTransaction.add(myBrokerClientFragment, "MyBroker");
                aTransaction.commit();
            }
            else
            {
                // This is not the first time, so it is the restart.
                // Therefore check the connection with the broker.
                if (!myBrokerClientFragment.isConnected())
                {
                    myBrokerClientFragment.recoverConnection();
                }
            }
            
            // Display the connection status.
            displayConnectionStatus(myBrokerClientFragment.isConnected());
            
            // Register to be informed about the connection.
            myBrokerClientFragment.connectionStatusChanged().subscribe(myOnConnectionChanged);
            
            // Register to be informed about the notification messages
            // received from the broker.
            myBrokerClientFragment.notifyMessageReceived().subscribe(myOnBrokerMessageReceived);
                
            // Subscribe to handle clicks on UI buttons.
            aSubscribe1Btn.setOnClickListener(mySubscribe1BtnClick);
            anUnsubscribe1Btn.setOnClickListener(myUnsubscribe1BtnClick);
            aSubscribe2Btn.setOnClickListener(mySubscribe2BtnClick);
            anUnsubscribe2Btn.setOnClickListener(myUnsubscribe2BtnClick);
            aSubscribe3Btn.setOnClickListener(mySubscribe3BtnClick);
            anUnsubscribe3Btn.setOnClickListener(myUnsubscribe3BtnClick);
            aReconnectBtn.setOnClickListener(myReconnectBtnClick);
        }
        catch (Exception err)
        {
            EneterTrace.error("Creating of activity failed.", err);
        }
    }
    
    // It is called when the Activity is destroyed or restarted.
    // e.g. when the device changed the screen orientation.
    @Override
    public void onDestroy()
    {
        // This activity ends so we need to unregister this activity
        // from events pushed from the broker fragment.
        // Note: If not unsubscribed then after restart the fragment would push events also
        //       into this already destroyed activity.
        myBrokerClientFragment.notifyMessageReceived().unsubscribe(myOnBrokerMessageReceived);
        myBrokerClientFragment.connectionStatusChanged().unsubscribe(myOnConnectionChanged);
        
        super.onDestroy();
    } 
    
    // Processes notification messages received from the broker.
    private void onBrokerMessageReceived(Object sender, final BrokerMessageReceivedEventArgs e)
    {
        if (e.getReceivingError() == null)
        {
            // Display the message using the UI thread.
            myRefresh.post(new Runnable()
            {
                @Override
                public void run()
                {
                    try
                    {
                        // Process notifications.
                        if (e.getMessageTypeId().equals("MyNotifyMsg1"))
                        {
                            NotifyMsg1 aMessage = mySerializer.deserialize(e.getMessage(), NotifyMsg1.class);
                            myMessage1EditText.setText(aMessage.CurrentTime);
                        }
                        else if (e.getMessageTypeId().equals("MyNotifyMsg2"))
                        {
                            NotifyMsg2 aMessage = mySerializer.deserialize(e.getMessage(), NotifyMsg2.class);
                            myMessage2EditText.setText(Integer.toString(aMessage.Number));
                        }
                        else if (e.getMessageTypeId().equals("MyNotifyMsg3"))
                        {
                            NotifyMsg3 aMessage = mySerializer.deserialize(e.getMessage(), NotifyMsg3.class);
                            myMessage3EditText.setText(aMessage.TextMessage);
                        }
                    }
                    catch (Exception err)
                    {
                        EneterTrace.error("Processing message from the broker failed.", err);
                    }
                }
            });
        }
    }
    
    // Subscribes for specified notification message in the broker.
    private void subscribe(String eventType)
    {
        try
        {
            // Send request to the broker to subscribe.
            myBrokerClientFragment.subscribeInBroker(eventType);
        }
        catch (Exception err)
        {
            EneterTrace.error("Subscribing to broker failed.", err);
        }
    }
    
    // Unsubscribes specified notification message from the broker.
    private void unsubscribe(String eventType)
    {
        try
        {
            // Send request to the broker to unsubscribe.
            myBrokerClientFragment.unsubscribeFromBroker(eventType);
        }
        catch (Exception err)
        {
            EneterTrace.error("Unsubscribing from broker failed.", err);
        }
    }
    
    // Displays the connection status.
    private void displayConnectionStatus(boolean isConnected)
    {
        // Get the connection status.
        final String aStatus = isConnected ? "Connected" : "Disconnected";
        
        // This event can come from various threads, so
        // enforce using the UI thread for displaying.
        myRefresh.post(new Runnable()
            {
                @Override
                public void run()
                {
                    myConnectionStatusTextView.setText(aStatus);
                }
            });
    }
    
    // Handler processing connection changes.
    private EventHandler<Boolean> myOnConnectionChanged =
            new EventHandler<Boolean>()
    {
        @Override
        public void onEvent(Object sender, Boolean e)
        {
            displayConnectionStatus(e);
        }
    };
    
    // Helper processing notifications received from the broker.
    private EventHandler<BrokerMessageReceivedEventArgs> myOnBrokerMessageReceived =
            new EventHandler<BrokerMessageReceivedEventArgs>()
    {
        @Override
        public void onEvent(Object sender, BrokerMessageReceivedEventArgs e)
        {
            onBrokerMessageReceived(sender, e);
        }
    };
    
    private OnClickListener mySubscribe1BtnClick = new OnClickListener()
    {
        @Override
        public void onClick(View v)
        {
            subscribe("MyNotifyMsg1");
        }
    };
    
    private OnClickListener myUnsubscribe1BtnClick = new OnClickListener()
    {
        @Override
        public void onClick(View v)
        {
            myMessage1EditText.setText("");
            unsubscribe("MyNotifyMsg1");
        }
    };
    
    private OnClickListener mySubscribe2BtnClick = new OnClickListener()
    {
        @Override
        public void onClick(View v)
        {
            subscribe("MyNotifyMsg2");
        }
    };
    
    private OnClickListener myUnsubscribe2BtnClick = new OnClickListener()
    {
        @Override
        public void onClick(View v)
        {
            myMessage2EditText.setText("");
            unsubscribe("MyNotifyMsg2");
        }
    };
    
    private OnClickListener mySubscribe3BtnClick = new OnClickListener()
    {
        @Override
        public void onClick(View v)
        {
            subscribe("MyNotifyMsg3");
        }
    };
    
    private OnClickListener myUnsubscribe3BtnClick = new OnClickListener()
    {
        @Override
        public void onClick(View v)
        {
            myMessage3EditText.setText("");
            unsubscribe("MyNotifyMsg3");
        }
    };
    
    private OnClickListener myReconnectBtnClick = new OnClickListener()
    {
        @Override
        public void onClick(View v)
        {
            myBrokerClientFragment.recoverConnection();
        }
    };
}

And here are applications communicating together. Three publishers sending notification messages to the broker that forwards them to the subscribed Android application. The whole communication is realized via webscokets.