<?xml version='1.0' encoding='UTF-8'?>
<!DOCTYPE chapter PUBLIC "-//OASIS//DTD DocBook XML V4.5//EN" "http://www.oasis-open.org/docbook/xml/4.5/docbookx.dtd" [
]>
        
<chapter>
    <title>Writing a simple application</title>

    <para>
        The goal of this chapter is to write a simple text-based chat application (SimpleChat), with the following features:
        <itemizedlist>
            <listitem>
                All instances of SimpleChat find each other and form a cluster.
            </listitem>
            <listitem>
                There is no need to run a central chat server to which instances have to connect. Therefore, there is no
                single point of failure.
            </listitem>
            <listitem>
                A chat message is sent to all instances of the cluster.
            </listitem>
            <listitem>
                An instance gets a notification callback when another instance leaves (or crashes) and when other
                instances join.
            </listitem>
            <listitem>
                (Optional) We maintain a common cluster-wide shared state, e.g. the chat history. New instances acquire
                that history from existing instances.
            </listitem>
        </itemizedlist>
    </para>

    <section>
        <title>JGroups overview</title>
        <para>
            JGroups uses a JChannel as the main API to connect to a cluster, send and receive messages, and to register
            listeners that are called when things (such as member joins) happen.
        </para>
        <para>
            What is sent around are Messages, which contain a byte buffer (the payload), plus the sender's and
            receiver's address. Addresses are subclasses of org.jgroups.Address, and usually contain an IP address plus
            a port.
        </para>
        <para>
            The list of instances in a cluster is called a View, and every instance contains
            exactly the same View. The list of the addresses of all instances can get retrieved by calling
            View.getMembers().
        </para>
        <para>
            Instances can only send or receive messages after they've joined a cluster.
        </para>
        <para>
            When an instance wants to leave the cluster, methods JChannel.disconnect() or JChannel.close() can be called.
            The latter actually calls disconnect() if the channel is still connected before closing the channel.
        </para>
    </section>

    <section>
        <title>Creating a channel and joining a cluster</title>
        <para>
            To join a cluster, we'll use a JChannel. An instance of JChannel is created with a configuration
            (e.g. an XML file) which defines the properties of the channel. To actually connect to the cluster, the
            connect(String clustername) method is used. All channel instances which call connect() with the same argument will
            join the same cluster. So, let's actually create a JChannel and connect to a cluster called "ChatCluster":
        </para>
        <programlisting language="Java">
import org.jgroups.JChannel;

public class SimpleChat {
    JChannel channel;
    String user_name=System.getProperty("user.name", "n/a");

    private void start() throws Exception {
        channel=new JChannel(); // use the default config, udp.xml
        channel.connect("ChatCluster");
    }

    public static void main(String[] args) throws Exception {
        new SimpleChat().start();
    }
}
        </programlisting>
        <para>
            First we create a channel using the empty constructor. This configures the channel with the default properties.
            Alternatively, we could pass an XML file to configure the channel, e.g. new JChannel("/home/bela/udp.xml").
        </para>
        <para>
            The connect() method joins cluster "ChatCluster". Note that we don't need to explicitly create a cluster
            beforehand; connect() creates the cluster if it is the first instance. All instances which join the same
            cluster will be in the same cluster, for example if we have
            <itemizedlist>
                <listitem>ch1 joining "cluster-one"</listitem>
                <listitem>ch2 joining "cluster-two"</listitem>
                <listitem>ch3 joining "cluster-two"</listitem>
                <listitem>ch4 joining "cluster-one"</listitem>
                <listitem>ch5 joining "cluster-three"</listitem>
            </itemizedlist>
            , then we will have 3 clusters: "cluster-one" with instances ch1 and ch4, "cluster-two" with ch2 and ch3,
            and "cluster-three" with only ch5.
        </para>
    </section>

    <section>
        <title>The main event loop and sending chat messages</title>
        <para>
            We now run an event loop, which reads input from stdin ('a message') and sends it to all
            instances currently in the cluster. When "exit" or "quit" are entered, we fall out of the
            loop and close the channel.
        </para>
        <programlisting language="Java">
private void start() throws Exception {
    channel=new JChannel();
    channel.connect("ChatCluster");
    eventLoop();
    channel.close();
}

private void eventLoop() {
    BufferedReader in=new BufferedReader(new InputStreamReader(System.in));
    while(true) {
        try {
            System.out.print("&gt; "); System.out.flush();
            String line=in.readLine().toLowerCase();
            if(line.startsWith("quit") || line.startsWith("exit"))
                break;
            line="[" + user_name + "] " + line;
            Message msg=new Message(null, null, line);
            channel.send(msg);
        }
        catch(Exception e) {
        }
    }
}
        </programlisting>
        <para>
            We added the call to eventLoop() and the closing of the channel to the start() method, and we provided an
            implementation of eventLoop.
        </para>
        <para>
            The event loop blocks until a new line is ready (from standard input), then sends a message to the cluster.
            This is done by creating a new Message and calling Channel.send() with it as argument.
        </para>
        <para>
            The first argument of the Message constructor is the destination address. A null destination address
            sends the message to everyone in the cluster (a non-null address of an instance would send the message
            to only one instance).
        </para>
        <para>
            The second argument is our own address. This is null as well, as the stack will insert the
            correct address anyway.
        </para>
        <para>
            The third argument is the line that we read from stdin, this uses Java
            serialization to create a byte[] buffer and set the message's payload. Note that we could also
            serialize the object ourselves (which is actually the recommended way !) and use the Message contructor which
            takes a byte[] buffer as third argument.
        </para>
        <para>
            The application is now fully functional, except that we don't yet receive messages or view notifications.
            This is done in the next section below.
        </para>
    </section>

    <section>
        <title>Receiving messages and view change notifications</title>
        <para>
            Let's now register as a Receiver to receive message and view changes. To this end, we could implement
            org.jgroups.Receiver, however, I chose to extend ReceiverAdapter which has default
            implementations, and only override callbacks (receive() and viewChange()) we're interested in. We
            now need to extend ReceiverAdapter:
        </para>

        <programlisting language="Java">public class SimpleChat extends ReceiverAdapter {</programlisting>

        <para>
            , set the receiver in start():
        </para>

                <programlisting language="Java">
private void start() throws Exception {
    channel=new JChannel();
    channel.setReceiver(this);
    channel.connect("ChatCluster");
    eventLoop();
    channel.close();
}
                </programlisting>

        <para>
            , and implement receive() and viewAccepted():
        </para>

        <programlisting language="Java">
public void viewAccepted(View new_view) {
    System.out.println("** view: " + new_view);
}

public void receive(Message msg) {
    System.out.println(msg.getSrc() + ": " + msg.getObject());
}
        </programlisting>

        <para>
            The viewAccepted() callback is called whenever a new instance joins the cluster, or an existing instance
            leaves (crashes included). Its toString() method prints out the view ID (an increasing ID) and a list of
            the current instances in the cluster
        </para>
        <para>
            In receive(), we get a Message as argument. We simply get its buffer as an object (again using Java
            serialization) and print it to stdout. We also print the sender's address (Message.getSrc()).
        </para>
        <para>
            Note that we could also get the byte[] buffer (the payload) by calling Message.getBuffer() and then
            de-serializing it ourselves, e.g. String line=new String(msg.getBuffer()).
        </para>
    </section>


    <section>
        <title>Trying out the SimpleChat application</title>
        <para>
            Now that the demo chat application is fully functional, let's try it out. Start an instance of SimpleChat:
            <screen>
[linux]/home/bela$ java SimpleChat

-------------------------------------------------------------------
GMS: address=linux-48776, cluster=ChatCluster, physical address=192.168.1.5:42442
-------------------------------------------------------------------
** view: [linux-48776|0] [linux-48776]
>
            </screen>
        </para>
        <para>
            The name of this instance is linux-48776 and the physical address is 192.168.1.5:42442 (IP address:port).
            A name is generated by JGroups (using the hostname and a random short) if the user doesn't set it. The name
            stays with an instance for its lifetime, and maps to an underlying UUID. The UUID then maps to a
            physical address.
        </para>
        <para>
            We started the first instance, let's start the second instance:
        </para>
        <para>
            <screen>
[linux]/home/bela$ java SimpleChat

-------------------------------------------------------------------
GMS: address=linux-37238, cluster=ChatCluster, physical address=192.168.1.5:40710
-------------------------------------------------------------------
** view: [linux-48776|1] [linux-48776, linux-37238]
>
            </screen>
        </para>
        <para>
            The cluster list is now [linux-48776, linux-37238], showing the first and second instance that
            joined the cluster. Note that the first instance (linux-48776) also received the same view, so both
            instances have the exact same view with the same ordering of its instances in the list. The instances are
            listed in order of joining the cluster, with the oldest instance as first element.
        </para>
        <para>
            Sending messages is now as simple as typing a message after the prompt and pressing return. The message will
            be sent to the cluster and therefore it will be received by both instances, including the sender.
        </para>
        <para>
            When "exit" or "quit" is entered, then the instance will leave the cluster. This means, a new
            view will be installed immediately.
        </para>
        <para>
            To simulate a crash, simply kill an instance (e.g. via CTRL-C, or from the process manager). The other
            surviving instance will receive a new view, with only 1 instance (itself) and excluding the crashed
            instance.
        </para>
    </section>

    <section>
        <title>Extra credits: maintaining shared cluster state</title>
        <para>
            One of the use cases of JGroups is to maintain state that is replicated across a cluster. For example, state
            could be all the HTTP sessions in a web server. If those sessions are replicated across a cluster, then
            clients can access any server in the cluster after a server which hosted the client's session crashed,
            and the user sessions will still be available.
        </para>
        <para>
            Any update to a session is replicated across the cluster, e.g. by serializing the attribute that was
            modified and sending the modification to every server in the cluster via JChannel.send(). This is needed
            so that all servers have the same state.
        </para>
        <para>
            However, what happens when a new server is started ? That server has to somehow get the state
            (e.g. all HTTP sessions) from an existing server in the cluster.
            This is called <emphasis>state transfer</emphasis>.
        </para>
        <para>
            State transfer in JGroups is done by implementing 2 (getState() and setState()) callbacks and
            calling the JChannel.getState() method.
            Note that, in order to be able to use state transfer in an application, the protocol stack has
            to have a state transfer protocol (the default stack used by the demo app does).
        </para>
        <para>
            The start() method is now modified to include the call to JChannel.getState():
        </para>

        <programlisting language="Java">
private void start() throws Exception {
    channel=new JChannel();
    channel.setReceiver(this);
    channel.connect("ChatCluster");
    channel.getState(null, 10000);
    eventLoop();
    channel.close();
}
        </programlisting>

        <para>
            The getState() method's first argument is the target instance, and null means get the state from the first
            instance (the coordinator). The second argument is the timeout; here's we're willing to wait for 10 secs
            to transfer the state. If the state cannot be transferred within this time, then an exception will be
            thrown. 0 means wait forever.
        </para>
        <para>
            ReceiverAdapter defines a callback getState() which is called on an existing instance
            (usually the coordinator) to fetch the cluster state. In our demo application, we define the state to be
            the chat conversation. This is a simple list, to the tail of which we add every message we receive.
            (Note that this is probably not the best example for state, as this state always grows. As a workaround,
            we could have a bounded list, which is not done here though).
        </para>
        <para>
            The list is defined as an instance variable:
        </para>

        <programlisting language="Java">final List&lt;String&gt; state=new LinkedList&lt;String&gt;();</programlisting>

        <para>
            Of course, now we need to modify receive() to add each received message to our state:
        </para>

        <programlisting language="Java">
public void receive(Message msg) {
    String line=msg.getSrc() + ": " + msg.getObject();
    System.out.println(line);
    synchronized(state) {
        state.add(line);
    }
}
        </programlisting>


        <para>
            The getState() callback implementation is
        </para>

        <programlisting language="Java">
public void getState(OutputStream output) throws Exception {
    synchronized(state) {
        Util.objectToStream(state, new DataOutputStream(output));
    }
}
        </programlisting>

        <para>
            The getState() method is called in the <emphasis>state provider</emphasis>, ie. an existing instance, to
            return the shared cluster state. It is passed an output stream to which the state has to be written. Note
            that JGroups closes that stream automatically after the state has been written, even in the case of an
            exception, so the stream doesn't need to be closed.
        </para>
        <para>
            Since access to <code>state</code> may be concurrent, we synchronize it. Then we call Util.objectToStream()
            which is a JGroups utility method writing an object to an output stream.
        </para>
        <para>
            The setState() method is called on the <emphasis>state requester</emphasis>, ie. the instance
            which called JChannel.getState(). Its task is to read the state from the input stream and set it
            accordingly:
        </para>

        <programlisting language="Java">
public void setState(InputStream input) throws Exception {
    List&lt;String&gt; list;
    list=(List&lt;String&gt;)Util.objectFromStream(new DataInputStream(input));
    synchronized(state) {
        state.clear();
        state.addAll(list);
    }
    System.out.println(list.size() + " messages in chat history):");
    for(String str: list) {
        System.out.println(str);
    }
}
        </programlisting>

        <para>
            We again call a JGroups utility method (Util.objectFromStream()) to create an object from an input stream.
        </para>
        <para>
            Then we synchronize on <literal>state</literal>, and set its contents from the received state.
        </para>
        <para>
            We also print the number of messages in the received chat history to stdout. Note that this is not
            feasible with a large chat history, but - again - we could have a bounded chat history list.
        </para>
    </section>

    <section>
        <title>Conclusion</title>
        <para>
            In this tutorial, we showed how to create a channel, join and leave a cluster, send and receive messages,
            get notified of view changes and implement state transfer. This is the core functionality provided by
            JGroups through the <code>JChannel</code> and <code>Receiver</code> APIs.
        </para>
        <para>
            JGroups has two more areas that weren't covered: building blocks and the protocol stack.
        </para>
        <para>
            Building blocks are classes residing on top of a JChannel that provide a higher abstraction level, e.g.
            request-response correlators, cluster-wide method calls, replicated hashmaps and so forth.
        </para>
        <para>
            The protocol stack allows for complete customization of JGroups: protocols can be configured, removed,
            replaced, enhanced, or new protocols can be written and added to the stack.
        </para>
        <para>
            The code for SimpleChat can be found <ulink url="../code/SimpleChat.java">here</ulink>.
        </para>
        <para>
            Here are some links for further information about JGroups:
            <itemizedlist>
                <listitem>
                    SimpleChat code: <ulink url="../code/SimpleChat.java">SimpleChat.java</ulink>
                </listitem>
                <listitem>
                    JGroups web site: <ulink url="http://www.jgroups.org">http://www.jgroups.org</ulink>
                </listitem>
                <listitem>
                    Downloads: <ulink url="http://sourceforge.net/projects/javagroups/files/JGroups/">here</ulink>
                </listitem>
                <listitem>
                    JIRA bug tracking: <ulink url="http://jira.jboss.com/jira/browse/JGRP">http://jira.jboss.com/jira/browse/JGRP</ulink>
                </listitem>
                <listitem>
                    Mailing lists: <ulink url="http://sourceforge.net/mail/?group_id=6081">http://sourceforge.net/mail/?group_id=6081</ulink>
                </listitem>
            </itemizedlist>
        </para>
    </section>

</chapter>



