<?xml version="1.0" encoding="UTF-8"?>
<chapter id="user-channel">
    <title>API</title>

    <para>
        This chapter explains the classes available in JGroups that will be used by applications to build reliable group
        communication applications. The focus is on creating and using channels.
    </para>

    <para>
        Information in this document may not be up-to-date, but the nature of the classes in JGroups
        described here is the same. For the most up-to-date information refer to the Javadoc-generated documentation in
        the <filename>doc/javadoc</filename> directory.
    </para>

    <para>All of the classes discussed here are in the <classname>org.jgroups</classname> package unless
        otherwise mentioned.
    </para>

    <section id="UtilityClasses">
        <title>Utility classes</title>

        <para>The <classname>org.jgroups.util.Util</classname> class contains useful common functionality which
            cannot be assigned to any other package.
        </para>

        <section id="objectToByteBuffer">
            <title>objectToByteBuffer(), objectFromByteBuffer()</title>

            <para>The first method takes an object as argument and serializes it into a byte buffer (the object has to
                be serializable or externalizable). The byte array is then returned. This method is often used to
                serialize objects into the byte buffer of a message. The second method returns a reconstructed object
                from a buffer. Both methods throw an exception if the object cannot be serialized or unserialized.
            </para>
        </section>
        <section id="objectToStream">
            <title>objectToStream(), objectFromStream()</title>

            <para>
                The first method takes an object and writes it to an output stream. The second method takes an
                input stream and reads an object from it.
                Both methods throw an exception if the object cannot be serialized or unserialized.
            </para>
        </section>
    </section>

    <section id="Interfaces">
        <title>Interfaces</title>

        <para>These interfaces are used with some of the APIs presented below, therefore they are listed first.</para>

        <section id="MessageListener">
            <title>MessageListener</title>

            <para>
                The <classname>MessageListener</classname> interface below provides callbacks for message reception and
                for providing and setting the state:
            </para>

            <programlisting language="Java">
public interface MessageListener {
    void receive(Message msg);
    void getState(OutputStream output) throws Exception;
    void setState(InputStream input) throws Exception;
}
            </programlisting>


            <para>Method <methodname>receive()</methodname> is be called whenever a message is received. The
                <methodname>getState()</methodname> and <methodname>setState()</methodname>
                methods are used to fetch and set the group state (e.g. when joining). Refer to
                <xref linkend="StateTransfer"/> for a discussion of state transfer.
            </para>
        </section>

        <section id="MembershipListener">
            <title>MembershipListener</title>

            <para>
                The <classname>MembershipListener</classname> interface is similar to the
                <classname>MessageListener</classname> interface above: every time a new view, a suspicion message,
                or a block event is received, the corresponding method of the class implementing
                <classname>MembershipListener</classname>  will be called.
            </para>

            <programlisting language="Java">
public interface MembershipListener {
    public void viewAccepted(View new_view);
    public void suspect(Object suspected_mbr);
    public void block();
    public void unblock();
}
            </programlisting>


            <para>
                Oftentimes the only callback that needs to be implemented will be
                <methodname>viewAccepted()</methodname> which notifies the receiver that a new member has joined the
                group or that an existing member has left or crashed. The <methodname>suspect()</methodname>
                callback is invoked by JGroups whenever a member if suspected of having crashed, but not yet excluded
                <footnote>
                    <para>It could be that the member is suspected falsely, in which case the next view would still
                        contain the suspected member (there is no <methodname>unsuspect()</methodname> method
                    </para>
                </footnote>.
            </para>

            <para>
                The <methodname>block()</methodname> method is called to notify the member that it will soon be blocked
                sending messages. This is done by the FLUSH protocol, for example to ensure that nobody is sending
                messages while a state transfer or view installation is in progress. When block() returns, any thread
                sending messages will be blocked, until FLUSH unblocks the thread again, e.g. after the state has been
                transferred successfully.
            </para>

            <para>
                Therefore, block() can be used to send pending messages or complete some other work.
                Note that block() should be brief, or else the entire FLUSH protocol is blocked.
            </para>

            <para>
                The <methodname>unblock()</methodname>
                method is called to notify the member that the FLUSH protocol has completed and the member can resume
                sending messages. If the member did not stop sending messages on block(), FLUSH simply blocked them and
                will resume, so no action is required from a member. Implementation of the unblock() callback is
                optional.
            </para>

            <note id="UseOfReceiverAdapter">
                <title>Use of MessageListener and MembershipListener</title>
                <para>
                    Note that it is oftentimes simpler to extend ReceiverAdapter (see below) and implement the needed
                    callbacks than to implement all methods of both of these interfaces, as most callbacks are not needed.
                </para>
            </note>

        </section>

        <section id="Receiver">
            <title>Receiver</title>
            <programlisting language="Java">public interface Receiver extends MessageListener, MembershipListener;</programlisting>

            <para>
                A Receiver can be used to receive messages and view changes; receive() will be invoked as soon as a
                message has been received, and viewAccepted() will be called whenever a new view is installed.
            </para>
        </section>


        <section id="ReceiverAdapter">
            <title>ReceiverAdapter</title>

            <para>
                This class implements Receiver with no-op implementations. When implementing a callback, we can simply
                extend ReceiverAdapter and overwrite receive() in order to not having to implement all callbacks of the
                interface.
            </para>

            <para>
                <classname>ReceiverAdapter</classname> looks as follows:
            </para>

            <programlisting language="Java">
public class ReceiverAdapter implements Receiver {
    public void receive(Message msg) {}
    public void getState(OutputStream output) throws Exception {}
    public void setState(InputStream input) throws Exception {}
    public void viewAccepted(View view) {}
    public void suspect(Address mbr) {}
    public void block() {}
    public void unblock() {}
}
            </programlisting>

            <para>
                A ReceiverAdapter is the recommended way to implement callbacks.
            </para>
        </section>

        <warning>
            <title>Sending messages in callbacks</title>
            <para>
                Note that anything that could block should <emphasis>not</emphasis>
                be done in a callback. This includes sending of messages; if we have FLUSH on the stack, and send a
                message in a viewAccepted() callback, then the following happens: the FLUSH protocol blocks all
                (multicast) messages before installing a view, then installs the view, then unblocks. However,
                because installation of the view triggers the viewAccepted() callback, sending of messages inside of
                viewAccepted() will block. This in turn blocks the viewAccepted() thread, so the flush will never
                return !
            </para>
            <para>
                If we need to send a message in a callback, the sending should be done on a separate thread, or a
                timer task should be submitted to the timer.
            </para>
        </warning>

        <section id="ChannelListener">
            <title>ChannelListener</title>
            <programlisting language="Java">
public interface ChannelListener {
    void channelConnected(Channel channel);
    void channelDisconnected(Channel channel);
    void channelClosed(Channel channel);
}
            </programlisting>

            <para>
                A class implementing <classname>ChannelListener</classname> can use the
                <methodname>Channel.addChannelListener()</methodname>
                method to register with a channel to obtain information about state changes in a channel. Whenever a
                channel is closed, disconnected or opened, the corresponding callback will be invoked.
            </para>
        </section>

    </section>

    <section id="Address">
        <title>Address</title>

        <para>Each member of a group has an address, which uniquely identifies the member. The interface for such an
            address is Address, which requires concrete implementations to provide methods such as comparison and
            sorting of addresses. JGroups addresses have to implement the following interface:
        </para>

        <programlisting language="Java">
public interface Address extends Externalizable, Comparable, Cloneable {
    int size();
}
        </programlisting>


        <para>
            For marshalling purposes, size() needs to return the number of bytes an instance of an address implementation
            takes up in serialized form.
        </para>

        <para>
            <emphasis>Please never use implementations of Address directly; Address should always be used as an opaque
                identifier of a cluster node !
            </emphasis>
        </para>

        <para>
            Actual implementations of addresses are often generated by the bottommost protocol layer (e.g. UDP or
            TCP). This allows for all possible sorts of addresses to be used with JGroups.
        </para>

        <para>
            Since an address uniquely identifies a channel, and therefore a group member, it
            can be used to send messages to that group member, e.g. in Messages (see next section).
        </para>
        <para>
            The default implementation of Address is <classname>org.jgroups.util.UUID</classname>. It uniquely identifies
            a node, and when disconnecting and reconnecting to a cluster, a node is given a new UUID on reconnection.
        </para>
        <para>
            UUIDs are never shown directly, but are usually shown as a logical name (see <xref linkend="LogicalName"/>).
            This is a name given to a node either via the user or via JGroups, and its sole purpose is to make logging
            output a bit more readable.
        </para>
        <para>
            UUIDs maps to IpAddresses, which are IP addresses and ports. These are eventually used by the transport
            protocol to send a message.
        </para>
    </section>

    <section id="Message">
        <title>Message</title>

        <para>
            Data is sent between members in the form of messages (<classname>org.jgroups.Message</classname>).
            A message can be sent by a member to a <emphasis>single member</emphasis>, or to
            <emphasis>all members</emphasis> of the group of which the channel is an endpoint.
            The structure of a message is shown in <xref linkend="MessageFig"/>.
        </para>

        <figure id="MessageFig">
            <title>Structure of a message</title>
            <graphic align="center" fileref="images/Message.png" format="PNG" scale="75"/>
        </figure>

        <para>A message has 5 fields:</para>

        <variablelist>
            <varlistentry>
                <term>Destination address</term>

                <listitem>
                    <para>
                        The address of the receiver. If <literal>null</literal>, the message will be sent to all
                        current group members. Message.getDest() returns the destination address of a message.
                    </para>
                </listitem>
            </varlistentry>

            <varlistentry>
                <term>Source address</term>

                <listitem>
                    <para>
                        The address of the sender. Can be left <literal>null</literal>, and will be filled in by the
                        transport protocol (e.g. UDP) before the message is put on the network.
                        Message.getSrc() returns the source address, ie. the address of the sender of a message.
                    </para>
                </listitem>
            </varlistentry>

            <varlistentry>
                <term>Flags</term>

                <listitem>
                    <para>
                        This is one byte used for flags. The currently recognized flags are OOB, DONT_BUNDLE, NO_FC,
                        NO_RELIABILITY, NO_TOTAL_ORDER, NO_RELAY and RSVP. For OOB, see the discussion on the concurrent stack
                        (<xref linkend="ConcurrentStack"/>). For the use of flags see <xref linkend="MessageFlags"/>.
                    </para>
                </listitem>
            </varlistentry>

            <varlistentry>
                <term>Payload</term>

                <listitem>
                    <para>The actual data (as a byte buffer). The Message class contains convenience methods to set a
                        serializable object and to retrieve it again, using serialization to convert the object to/from
                        a byte buffer. A message also has an offset and a length, if the buffer is only a subrange
                        of a larger buffer.
                    </para>
                </listitem>
            </varlistentry>

            <varlistentry>
                <term>Headers</term>

                <listitem>
                    <para>A list of headers that can be attached to a message. Anything that should not be in the
                        payload can be attached to a message as a header. Methods <methodname>putHeader()</methodname>
                        , <methodname>getHeader()</methodname> and <methodname>removeHeader()</methodname>
                        of Message can be used to manipulate headers.
                    </para>
                    <para>
                        Note that headers are only used by protocol implementers; headers should not be added or
                        removed by application code !
                    </para>
                </listitem>
            </varlistentry>
        </variablelist>

        <para>
            A message is similar to an IP packet and consists of the payload (a byte buffer) and the addresses of the
            sender and receiver (as Addresses). Any message put on the network can be routed to its destination
            (receiver address), and replies can be returned to the sender's address.
        </para>

        <para>
            A message usually does not need to fill in the sender's address when sending a message; this is done
            automatically by the protocol stack before a message is put on the network. However, there may be cases,
            when the sender of a message wants to give an address different from its own, so that for example, a
            response should be returned to some other member.
        </para>

        <para>
            The destination address (receiver) can be an Address, denoting the address of a member, determined e.g.
            from a message received previously, or it can be <literal>null</literal>, which means that the message
            will be sent to all members of the group. A typical multicast message, sending string
            <literal>"Hello"</literal> to all members would look like this:
        </para>

        <programlisting language="Java">
Message msg=new Message(null, "Hello");
channel.send(msg);
        </programlisting>

    </section>

    <section id="Header">
        <title>Header</title>

        <para>
            A header is a custom bit of information that can be added to each message. JGroups uses headers extensively,
            for example to add sequence numbers to each message (NAKACK and UNICAST), so that those messages can be
            delivered in the order in which they were sent.
        </para>

    </section>

    <section id="Event">
        <title>Event</title>

        <para>
            Events are means by which JGroups protcols can talk to each other. Contrary to Messages, which travel over
            the network between group members, events only travel up and down the stack.
        </para>
        <note>
            <title>Headers and events</title>
            <para>
                Headers and events are only used by protocol implementers; they are not needed by application code !
            </para>
        </note>

    </section>


    <section id="View">
        <title>View</title>

        <para>
            A view (<classname>org.jgroups.View</classname>) is a list of the current members of a group. It consists
            of a <classname>ViewId</classname>, which uniquely identifies the view (see below), and a list of members.
            Views are installed in a channel automatically by the underlying protocol stack whenever a new member joins
            or an existing one leaves (or crashes). All members of a group see the same sequence of views.
        </para>

        <para>
            Note that the first member of a view is the <emphasis>coordinator</emphasis> (the one who emits new views).
            Thus, whenever the membership changes, every member can determine the coordinator easily and without having
            to contact other members, by picking the first member of a view.
        </para>

        <para>
            The code below shows how to send a (unicast) message to the first member of a view (error checking code
            omitted):
        </para>

        <programlisting language="Java">
View view=channel.getView();
Address first=view.getMembers().get(0);
Message msg=new Message(first, "Hello world");
channel.send(msg);
        </programlisting>


        <para>
            Whenever an application is notified that a new view has been installed (e.g. by
            <methodname>Receiver.viewAccepted()</methodname>, the view is already set in the channel. For example,
            calling <methodname>Channel.getView()</methodname> in a <methodname>viewAccepted()</methodname>
            callback would return the same view (or possibly the next one in case there has already been a new view !).
        </para>

        <section id="ViewId">
            <title>ViewId</title>

            <para>The ViewId is used to uniquely number views. It consists of the address of the view creator and a
                sequence number. ViewIds can be compared for equality and put in a hashmaps as they implement equals()
                and hashCode().
                <footnote><para>Note that the latter 2 methods only take the ID into account.</para></footnote>
            </para>
        </section>

        <section id="MergeView">
            <title>MergeView</title>

            <para>
                Whenever a group splits into subgroups, e.g. due to a network partition, and later the subgroups merge
                back together, a MergeView instead of a View will be received by the application. The MergeView is
                a subclass of View and contains as additional instance variable the list of views that were merged. As
                an example if the group denoted by view <literal>V1:(p,q,r,s,t)</literal> split into subgroups
                <literal>V2:(p,q,r)</literal> and <literal>V2:(s,t)</literal>, the merged view might be
                <literal>V3:(p,q,r,s,t)</literal>. In this case the MergeView would contains a list of 2 views:
                <literal>V2:(p,q,r)</literal> and <literal>V2:(s,t)</literal>.
            </para>
        </section>
    </section>


    <section id="JChannel">
        <title>JChannel</title>

        <para>
            In order to join a group and send messages, a process has to create a channel. A channel is like a socket.
            When a client connects to a channel, it gives the the name of the group it would like to join. Thus, a
            channel is (in its connected state) always associated with a particular group. The protocol stack takes care
            that channels with the same group name find each other: whenever a client connects to a channel given group
            name G, then it tries to find existing channels with the same name, and joins them, resulting in a new view
            being installed (which contains the new member). If no members exist, a new group will be created.
        </para>

        <para>
            A state transition diagram for the major states a channel can assume are shown in
            <xref linkend="ChannelStatesFig"/>.
        </para>

        <figure id="ChannelStatesFig">
            <title>Channel states</title>
            <graphic align="center" fileref="images/ChannelStates.png" format="PNG" scale="75"/>
        </figure>

        <para>
            When a channel is first created, it is in the unconnected state. An attempt to perform certain operations
            which are only valid in the connected state (e.g. send/receive messages) will result in an exception. After
            a successful connection by a client, it moves to the connected state. Now the channel will receive messages
            from other members and may send messages to other members or to the group, and it will get notified when
            new members join or leave. Getting the local address of a channel is guaranteed to be a valid operation in
            this state (see below). When the channel is disconnected, it moves back to the unconnected state.
            Both a connected and unconnected channel may be closed, which makes the channel unusable for further
            operations. Any attempt to do so will result in an exception. When a channel is closed directly
            from a connected state, it will first be disconnected, and then closed.
        </para>

        <para>
            The methods available for creating and manipulating channels are discussed now.
        </para>

        <section id="CreatingAChannel">
            <title>Creating a channel</title>

            <para>
                A channel is created using one of its public constructors (e.g. <methodname>new JChannel()</methodname>).
            </para>

            <para>
                The most frequently used constructor of <classname>JChannel</classname> looks as follows:
            </para>

            <programlisting language="Java">public JChannel(String props) throws Exception;</programlisting>

            <para>
                The <parameter>props</parameter> argument points to an XML file containing the configuration of the
                protocol stack to be used. This can be a String, but there are also other constructors which take for
                example a DOM element or a URL (see the javadoc for details).
            </para>

            <para>
                The code sample below shows how to create a channel based on an XML configuration file:
            </para>

            <programlisting language="Java">JChannel ch=new JChannel("/home/bela/udp.xml");</programlisting>

            <para>
                If the props argument is null, the default properties will be used. An exception will be thrown if the
                channel cannot be created. Possible causes include protocols that were specified in the property
                argument, but were not found, or wrong parameters to protocols.
            </para>

            <para>For example, the Draw demo can be launched as follows:</para>

            <screen>java org.javagroups.demos.Draw -props file:/home/bela/udp.xml</screen>

            <para>or</para>

            <screen>java org.javagroups.demos.Draw -props http://www.jgroups.org/udp.xml</screen>

            <para>
                In the latter case, an application downloads its protocol stack specification from a server, which
                allows for central administration of application properties.
            </para>
            <para>
                A sample XML configuration looks like this (edited from udp.xml):
            </para>
            <programlisting language="XML">
&lt;config xmlns="urn:org:jgroups"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:schemaLocation="urn:org:jgroups http://www.jgroups.org/schema/JGroups-3.0.xsd"&gt;
    &lt;UDP
         mcast_port="${jgroups.udp.mcast_port:45588}"
         tos="8"
         ucast_recv_buf_size="20M"
         ucast_send_buf_size="640K"
         mcast_recv_buf_size="25M"
         mcast_send_buf_size="640K"
         loopback="true"
         discard_incompatible_packets="true"
         max_bundle_size="64K"
         max_bundle_timeout="30"
         ip_ttl="${jgroups.udp.ip_ttl:2}"
         enable_bundling="true"
         enable_diagnostics="true"
         thread_naming_pattern="cl"

         timer_type="new"
         timer.min_threads="4"
         timer.max_threads="10"
         timer.keep_alive_time="3000"
         timer.queue_max_size="500"

         thread_pool.enabled="true"
         thread_pool.min_threads="2"
         thread_pool.max_threads="8"
         thread_pool.keep_alive_time="5000"
         thread_pool.queue_enabled="true"
         thread_pool.queue_max_size="10000"
         thread_pool.rejection_policy="discard"

         oob_thread_pool.enabled="true"
         oob_thread_pool.min_threads="1"
         oob_thread_pool.max_threads="8"
         oob_thread_pool.keep_alive_time="5000"
         oob_thread_pool.queue_enabled="false"
         oob_thread_pool.queue_max_size="100"
         oob_thread_pool.rejection_policy="Run"/&gt;

    &lt;PING timeout="2000"
            num_initial_members="3"/&gt;
    &lt;MERGE2 max_interval="30000"
            min_interval="10000"/&gt;
    &lt;FD_SOCK/&gt;
    &lt;FD_ALL/&gt;
    &lt;VERIFY_SUSPECT timeout="1500"  /&gt;
    &lt;BARRIER /&gt;
    &lt;pbcast.NAKACK use_stats_for_retransmission="false"
                   exponential_backoff="0"
                   use_mcast_xmit="true"
                   retransmit_timeout="300,600,1200"
                   discard_delivered_msgs="true"/&gt;
    &lt;UNICAST timeout="300,600,1200"/&gt;
    &lt;pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000"
                   max_bytes="4M"/&gt;
    &lt;pbcast.GMS print_local_addr="true" join_timeout="3000"
                view_bundling="true"/&gt;
    &lt;UFC max_credits="2M"
         min_threshold="0.4"/&gt;
    &lt;MFC max_credits="2M"
         min_threshold="0.4"/&gt;
    &lt;FRAG2 frag_size="60K"  /&gt;
    &lt;pbcast.STATE_TRANSFER /&gt;
&lt;/config>
            </programlisting>

            <para>
                A stack is wrapped by &lt;config&gt; and &lt;/config&gt; elements and lists all protocols from bottom
                (UDP) to top (STATE_TRANSFER). Each element defines one protocol.
            </para>

            <para>
                Each protocol is implemented as a Java class. When a protocol stack is created based on the above XML
                configuration, the first element ("UDP") becomes the bottom-most layer, the second one will be placed on
                the first, etc: the stack is created from the bottom to the top.
            </para>

            <para>
                Each element has to be the name of a Java class that resides in the <classname>org.jgroups.protocols</classname>
                package. Note that only the base name has to be given, not the fully specified class name (
                <classname>UDP</classname> instead of <classname>org.jgroups.protocols.UDP</classname>).
                If the protocol class is not found, JGroups assumes that the name given is a fully qualified classname
                and will therefore try to instantiate that class. If this does not work an exception is thrown.
                This allows for protocol classes to reside in different packages altogether, e.g. a valid protocol name
                could be <classname>com.sun.eng.protocols.reliable.UCAST</classname>.
            </para>

            <para>
                Each layer may have zero or more arguments, which are specified as a list of name/value pairs in
                parentheses directly after the protocol name. In the example above, UDP is configured with some options,
                one of them being the IP multicast port (mcast_port) which is set to 45588, or to the value of
                the system property jgroups.udp.mcast_port, if set.
            </para>

            <para>
                <emphasis>Note that all members in a group have to have the same protocol stack.</emphasis>
            </para>


            <section id="ProgrammaticCreation">
                <title>Programmatic creation</title>
                <para>
                    Usually, channels are created by passing the name of an XML configuration file to the JChannel()
                    constructor. On top of this declarative configuration, JGroups provides an API to create a channel
                    programmatically. The way to do this is to first create a JChannel, then an instance of
                    ProtocolStack, then add all desired protocols to the stack and finally calling init() on the stack
                    to set it up. The rest, e.g. calling JChannel.connect() is the same as with the declarative
                    creation.
                </para>
                <para>
                    An example of how to programmatically create a channel is shown below (copied from
                    ProgrammaticChat):
                </para>
                    <programlisting language="Java">
public class ProgrammaticChat {

    public static void main(String[] args) throws Exception {
        JChannel ch=new JChannel(false);         // (1)
        ProtocolStack stack=new ProtocolStack(); // (2)
        ch.setProtocolStack(stack);
        stack.addProtocol(new UDP().setValue("bind_addr",
                                              InetAddress.getByName("192.168.1.5")))
                .addProtocol(new PING())
                .addProtocol(new MERGE2())
                .addProtocol(new FD_SOCK())
                .addProtocol(new FD_ALL().setValue("timeout", 12000)
                                         .setValue("interval", 3000))
                .addProtocol(new VERIFY_SUSPECT())
                .addProtocol(new BARRIER())
                .addProtocol(new NAKACK())
                .addProtocol(new UNICAST2())
                .addProtocol(new STABLE())
                .addProtocol(new GMS())
                .addProtocol(new UFC())
                .addProtocol(new MFC())
                .addProtocol(new FRAG2());       // (3)
        stack.init();                            // (4)

        ch.setReceiver(new ReceiverAdapter() {
            public void viewAccepted(View new_view) {
                System.out.println("view: " + new_view);
            }

            public void receive(Message msg) {
                Address sender=msg.getSrc();
                System.out.println(msg.getObject() + " [" + sender + "]");
            }
        });

        ch.connect("ChatCluster");


        for(;;) {
            String line=Util.readStringFromStdin(": ");
            ch.send(null, line);
        }
    }

}
                    </programlisting>

                <para>
                    First a JChannel is created (1). The 'false' argument tells the channel not to create a ProtocolStack.
                    This is needed because we will create one ourselves later and set it in the channel (2).
                </para>
                <para>
                    Next, all protocols are added to the stack (3). Note that the order is from bottom (transport protocol)
                    to top. So UDP as transport is added first, then PING and so on, until FRAG2, which is the top
                    protocol. Every protocol can be configured via setters, but there is also a generic setValue(String
                    attr_name, Object value), which can be used to configure protocols as well, as shown in the example.
                </para>
                <para>
                    Once the stack is configured, we call ProtocolStack.init() to link all protocols correctly and to
                    call init() in every protocol instance (4). After this, the channel is ready to be used and all
                    subsequent actions (e.g. connect()) can be executed. When the init() method returns, we have
                    essentially the equivalent of new JChannel(config_file).
                </para>
            </section>


        </section>



        <section id="LogicalName">
            <title>Giving the channel a logical name</title>
            <para>
                A channel can be given a logical name which is then used instead of the channel's address in toString().
                A logical name might show the function of a channel, e.g. "HostA-HTTP-Cluster", which is more legible
                than a UUID <literal>3c7e52ea-4087-1859-e0a9-77a0d2f69f29</literal>.
            </para>
            <para>
                For example, when we have 3 channels, using logical names we might see a view "{A,B,C}", which is nicer
                than "{<literal>56f3f99e-2fc0-8282-9eb0-866f542ae437</literal>,
                <literal>ee0be4af-0b45-8ed6-3f6e-92548bfa5cde</literal>,
                <literal>9241a071-10ce-a931-f675-ff2e3240e1ad</literal>} !"
            </para>
            <para>
                If no logical name is set, JGroups generates one, using the hostname and a random number, e.g.
                linux-3442. If this is not desired and the UUIDs should be shown, use system property
                -Djgroups.print_uuids=true.
            </para>
            <para>
                The logical name can be set using:
            </para>
            <programlisting language="Java">public void setName(String logical_name);</programlisting>

            <para>
                This must be done <emphasis>before</emphasis> connecting a channel. Note that the logical name stays
                with a channel until the channel is destroyed, whereas a UUID is created on each connection.
            </para>

            <para>
                When JGroups starts, it prints the logical name and the associated physical address(es):
                <screen>
-------------------------------------------------------------------
GMS: address=mac-53465, cluster=DrawGroupDemo, physical address=192.168.1.3:49932
-------------------------------------------------------------------</screen>

                The logical name is mac-53465 and the physical address is 192.168.1.3:49932. The UUID is not shown here.
            </para>
        </section>

        <section id="CustomAddresses">
            <title>Generating custom addresses</title>
            <para>
                Since 2.12 address generation is pluggable. This means that an application can determine what kind of
                addresses it uses. The default address type is UUID, and since some protocols use UUID, it is
                recommended to provide custom classes as <emphasis>subclasses of UUID</emphasis>.
            </para>
            <para>
                This can be used to for example pass additional data around with an address, for example information
                about the location of the node to which the address is assigned. Note that methods equals(), hashCode()
                and compare() of the UUID super class should not be changed.
            </para>

            <para>
                To use custom addresses, an implementation of <classname>org.jgroups.stack.AddressGenerator</classname>
                has to be written.
            </para>
            <para>
                For any class CustomAddress, it will need to get registered with the ClassConfigurator in
                order to marshal it correctly:
            </para>
            <programlisting language="Java">
class CustomAddress extends UUID {
    static {
        ClassConfigurator.add((short)8900, CustomAddress.class);
    }
}
            </programlisting>

            <note>
                <para>
                    Note that the ID should be chosen such that it doesn't collide with any IDs defined in
                    jg-magic-map.xml.
                </para>
            </note>
            <para>
                Set the address generator in JChannel: setAddressGenerator(AddressGenerator). This has to
                be done <emphasis>before</emphasis> the channel is connected.
            </para>

            <para>
                An example of a subclass is <classname>org.jgroups.util.PayloadUUID</classname>, and there are
                2 more shipped with JGroups.
            </para>
        </section>

        <section id="JoiningACluster">
            <title>Joining a cluster</title>

            <para>
                When a client wants to join a cluster, it <emphasis>connects</emphasis>
                to a channel giving the name of the cluster to be joined:
            </para>

            <programlisting language="Java">public void connect(String cluster) throws Exception;</programlisting>

            <para>
                The cluster name is the name of the cluster to be joined. All channels that call connect() with
                the same name form a cluster. Messages sent on any channel in the cluster will be received by all
                members (including the one who sent it
                <footnote>
                    <para>
                        Local delivery can be turned off using <methodname>setDiscardOwnMessages(true)</methodname>.
                    </para>
                </footnote>
                ).
            </para>

            <para>
                The connect() method returns as soon as the cluster has been joined successfully. If the channel is in
                the closed state (see <xref linkend="ChannelStatesFig"/>), an exception will be thrown. If there are
                no other members, i.e. no other member has connected to a cluster with this name, then a new cluster is
                created and the member joins it as first member. The first member of a cluster becomes its
                <emphasis>coordinator</emphasis>.
                A coordinator is in charge of installing new views whenever the membership changes
                <footnote>
                    <para>This is managed internally however, and an application programmer does not need to be
                        concerned about it.
                    </para>
                </footnote>
                .
            </para>
        </section>

        <section id="AtomicJoinAndGetState">
            <title>Joining a cluster and getting the state in one operation</title>
            <para>
                Clients can also join a cluster and fetch cluster state <emphasis>in one operation</emphasis>.
                The best way to conceptualize the connect and fetch state connect method is to think of it as an
                invocation of the regular connect() and getState() methods executed in succession. However, there are
                several advantages of using the connect and fetch state connect method over the regular connect. First
                of all, the underlying message exchange is heavily optimized, especially if the flush protocol is used.
                But more importantly, from a client's perspective, the connect() and fetch state operations become
                one atomic operation.
            </para>

            <programlisting language="Java">public void connect(String cluster, Address target, long timeout) throws Exception;</programlisting>

            <para>
                Just as in a regular connect(), the cluster name represents a cluster to be joined. The target parameter
                indicates a cluster member to fetch the state from. A null target indicates that the state should be
                fetched from the cluster coordinator. If the state should be fetched from a particular member other than
                the coordinator, clients can simply provide the address of that member. The timeout paremeter bounds
                the entire join and fetch operation. An exception will be thrown if the timeout is exceeded.
            </para>
        </section>

        <section id="GettingAddress">
            <title>Getting the local address and the cluster name</title>

            <para>
                Method <methodname>getAddress()</methodname> returns the address of the channel. The address may or may
                not be available when a channel is in the unconnected state.
            </para>

            <programlisting language="Java">public Address getAddress();</programlisting>

            <para>
                Method <methodname>getClusterName()</methodname> returns the name of the cluster which the member joined.
            </para>

            <programlisting language="Java">public String getClusterName();</programlisting>

            <para>
                Again, the result is undefined if the channel is in the disconnected or closed state.
            </para>
        </section>

        <section id="GettingTheView">
            <title>Getting the current view</title>

            <para>The following method can be used to get the current view of a channel:</para>

            <programlisting language="Java">public View getView();</programlisting>

            <para>
                This method returns the current view of the channel. It is updated every time a new view is
                installed (viewAccepted() callback).
            </para>

            <para>
                Calling this method on an unconnected or closed channel is implementation defined. A channel may
                return null, or it may return the last view it knew of.
            </para>
        </section>

        <section id="SendingMessages">
            <title>Sending messages</title>

            <para>
                Once the channel is connected, messages can be sent using one of the <methodname>send()</methodname>
                methods:
            </para>

            <programlisting language="Java">
public void send(Message msg) throws Exception;
public void send(Address dst, Serializable obj) throws Exception;
public void send(Address dst, byte[] buf) throws Exception;
public void send(Address dst, byte[] buf, int off, int len) throws Exception;
            </programlisting>

            <para>
                The first <methodname>send()</methodname> method has only one argument, which is the message to be sent.
                The message's destination should either be the address of the receiver (unicast) or null (multicast).
                When the destination is null, the message will be sent to all members of the cluster (including itself).
            </para>

            <para>
                The remainaing <methodname>send()</methodname> methods are helper methods; they take either a byte[]
                buffer or a serializable, create a Message and call send(Message).
            </para>

            <para>
                If the channel is not connected, or was closed, an exception will be thrown upon attempting to send a
                message.
            </para>

            <para>
                Here's an example of sending a message to all members of a cluster:
            </para>

            <programlisting language="Java">
Map data; // any serializable data
channel.send(null, data);
            </programlisting>


            <para>
                The null value as destination address means that the message will be sent to all members in the cluster.
                The payload is a hashmap, which will be serialized into the message's buffer and unserialized at the
                receiver. Alternatively, any other means of generating a byte buffer and setting the message's buffer
                to it (e.g. using Message.setBuffer()) also works.
            </para>

            <para>
                Here's an example of sending a unicast message to the first member (coordinator) of a group:
            </para>

            <programlisting language="Java">
Map data;
Address receiver=channel.getView().getMembers().get(0);
channel.send(receiver, "hello world");
            </programlisting>


            <para>
                The sample code determines the coordinator (first member of the view) and sends it a "hello world"
                message.
            </para>

            <section id="DiscardingOwnMessages">
                <title>Discarding one's own messages</title>

                <para>
                    Sometimes, it is desirable not to have to deal with one's own messages, ie. messages sent by oneself.
                    To do this, <methodname>JChannel.setDiscardOwnMessages(boolean flag)</methodname> can be set to
                    true (false by default). This means that every cluster node will receive a message sent
                    by P, but P itself won't.
                </para>
                <para>
                    Note that this method replaces the old JChannel.setOpt(LOCAL, false) method, which was removed in 3.0.
                </para>
            </section>

            <section id="RsvpSection">
                <title>Synchronous messages</title>
                <para>
                    While JGroups guarantees that a message will eventually be delivered at all non-faulty members,
                    sometimes this might take a while. For example, if we have a retransmission protocol based on
                    negative acknowledgments, and the last message sent is lost, then the receiver(s) will have to
                    wait until the stability protocol notices that the message has been lost, before it can be
                    retransmitted.
                </para>
                <para>
                    This can be changed by setting the Message.RSVP flag in a message: when this flag is encountered,
                    the message send blocks until all members have acknowledged reception of the message (of course
                    excluding members which crashed or left meanwhile).
                </para>
                <para>
                    This also serves as another purpose: if we send an RSVP-tagged message, then - when the send()
                    returns - we're guaranteed that all messages sent <emphasis>before</emphasis> will have been
                    delivered at all members as well. So, for example, if P sends message 1-10, and marks 10 as RSVP,
                    then, upon JChannel.send() returning, P will know that all members received messages 1-10 from P.
                </para>
                <para>
                    Note that since RSVP'ing a message is costly, and might block the sender for a while, it should be
                    used sparingly. For example, when completing a unit of work (ie. member P sending N messages), and
                    P needs to know that all messages were received by everyone, then RSVP could be used.
                </para>
                <para>
                    To use RSVP, 2 things have to be done:
                </para>
                <para>
                    First, the RSVP protocol has to be in the config, somewhere above the reliable transmission
                    protocols such as NAKACK or UNICAST(2), e.g.:
                </para>

        <programlisting language="XML">
&lt;config&gt;
    &lt;UDP/&gt;
    &lt;PING /&gt;
    &lt;FD_ALL/&gt;
    &lt;pbcast.NAKACK use_mcast_xmit="true"
                   discard_delivered_msgs="true"/&gt;
    &lt;UNICAST timeout="300,600,1200"/&gt;
    &lt;RSVP /&gt;
    &lt;pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000"
                   max_bytes="4M"/&gt;
    &lt;pbcast.GMS print_local_addr="true" join_timeout="3000"
                view_bundling="true"/&gt;
    ...
&lt;/config&gt;
        </programlisting>

                <para>
                    Secondly, the message we want to get ack'ed must be tagged with RSVP:
                </para>
                <programlisting language="Java">
Message msg=new Message(null, null, "hello world");
msg.setFlag(Message.RSVP);
ch.send(msg);
                </programlisting>
                <para>
                    Here, we send a message to all cluster members (dest = null). (Note that RSVP also works for sending
                    a message to a unicast destination). Method send() will return as soon as it has received acks from
                    all current members. If there are 4 members A, B, C and D, and A has received acks from itself, B
                    and C, but D's ack is missing and D crashes before the timeout kicks in, then this will
                    nevertheless make send() return, as if D had actually sent an ack.
                </para>
                <para>
                    If the <code>timeout</code> property if greater than 0, and we don't receive all acks within
                    timeout milliseconds, a TimeoutException will be thrown (if RSVP.throw_exception_on_timeout is true).
                    The application can choose to catch this (runtime) exception and do something with it, e.g. retry.
                </para>
                <para>
                    The configuration of RSVP is described here: <xref linkend="RSVP"/>.
                </para>

                <note>
                    <para>
                        RSVP was added in version 3.1.
                    </para>
                </note>
            </section>
        </section>



        <section id="ReceivingMessages">
            <title>Receiving messages</title>

            <para>
                Method <methodname>receive()</methodname> in ReceiverAdapter (or Receiver) can be overridden to
                receive messages, views, and state transfer callbacks.
            </para>

            <programlisting language="Java">public void receive(Message msg);</programlisting>


            <para>
                A Receiver can be registered with a channel using JChannel.setReceiver(). All received messages, view
                changes and state transfer requests will invoke callbacks on the registered Receiver:
             </para>

            <programlisting language="Java">
JChannel ch=new JChannel();
ch.setReceiver(new ReceiverAdapter() {
    public void receive(Message msg) {
        System.out.println("received message " + msg);
    }
    public void viewAccepted(View view) {
        System.out.println("received view " + new_view);
    }
});
ch.connect("MyCluster");
            </programlisting>



        </section>

        <section id="ReceivingViewChanges">
            <title>Receiving view changes</title>
            <para>
                As shown above, the <methodname>viewAccepted()</methodname> callback of ReceiverAdapter can be used
                to get callbacks whenever a cluster membership change occurs. The receiver needs to be set via
                <methodname>JChannel.setReceiver(Receiver)</methodname>.
            </para>
            <para>
                As discussed in <xref linkend="ReceiverAdapter"/>, code in callbacks must avoid anything that takes
                a lot of time, or blocks; JGroups invokes this callback as part of the view installation, and if this
                user code blocks, the view installation would block, too.
            </para>
        </section>


        <section id="StateTransfer">
            <title>Getting the group's state</title>

            <para>
                A newly joined member may want to retrieve the state of the cluster before starting work. This is done
                with <methodname>getState()</methodname>:
            </para>
            <programlisting language="Java">public void getState(Address target, long timeout) throws Exception;</programlisting>

            <para>
                This method returns the state of one member (usually of the oldest member, the coordinator). The target
                parameter can usually be null, to ask the current coordinator for the state. If a timeout (ms) elapses
                before the state is fetched, an exception will be thrown. A timeout of 0 waits until the entire state
                has been transferred.
            </para>

            <note>
                <para>The reason for not directly returning the state as a result of
                    <methodname>getState()</methodname> is that the state has to be returned in the correct position
                    relative to other messages. Returning it directly would violate the FIFO properties of a channel,
                    and state transfer would not be correct !
                </para>
            </note>

            <para>
                To participate in state transfer, both state provider and state requester have to implement the
                following callbacks from ReceiverAdapter (Receiver):
            </para>
            
            <programlisting language="Java">
public void getState(OutputStream output) throws Exception;
public void setState(InputStream input) throws Exception;
            </programlisting>

            <para>
                Method getState() is invoked on the <emphasis>state provider</emphasis> (usually the coordinator). It
                needs to write its state to the output stream given. Note that output doesn't need to be closed when
                done (or when an exception is thrown); this is done by JGroups.
            </para>

            <para>
                The setState() method is invoked on the <emphasis>state requester</emphasis>; this is the member
                which called JChannel.getState(). It needs to read its state from the input stream and set its
                internal state to it. Note that input doesn't need to be closed when
                done (or when an exception is thrown); this is done by JGroups.
            </para>

            <para>
                In a cluster consisting of A, B and C, with D joining the cluster and calling Channel.getState(), the
                following sequence of callbacks happens:
                <itemizedlist>
                    <listitem>
                        D calls JChannel.getState(). The state will be retrieved from the oldest member, A
                    </listitem>

                    <listitem>
                        A's getState() callback is called. A writes its state to the output stream passed as a
                        parameter to getState().
                    </listitem>

                    <listitem>
                        D's setState() callback is called with an input stream as argument. D reads the state from the
                        input stream and sets its internal state to it, overriding any previous data.
                    </listitem>

                    <listitem>
                        D: JChannel.getState() returns. Note that this will only happen <emphasis>after</emphasis> the
                        state has been transferred successfully, or a timeout elapsed, or either the state provider or
                        requester throws an exception. Such an exception will be re-thrown by getState(). This could
                        happen for instance if the state provider's getState() callback tries to stream a
                        non-serializable class to the output stream.
                    </listitem>
                </itemizedlist>
            </para>

            

            <para>The following code fragment shows how a group member participates in state transfers:</para>

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

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>
                This code is the Chat example from the JGroups tutorial and the state here is a list of strings.
            </para>

            <para>
                The getState() implementation synchronized on the state (so no incoming messages can modify it during
                the state transfer), and uses the JGroups utility method objectToStream().
            </para>

            <note>
                <title>Performance when writing to an output stream</title>
                <para>
                    If a lot of smaller fragments are written to an output stream, it is best to wrap the output stream
                    into a BufferedOutputStream, e.g.
                </para>
                <programlisting language="Java">
Util.objectToStream(state,
                    new BufferedOutputStream(
                        new DataOutputStream(output)));
                </programlisting>
            </note>

            <para>
                The setState() implementation also uses the Util.objectFromStream() utility method to read the state from
                the input stream and assign it to its internal list.
            </para>


            <section id="StateTransferProtocols">
                <title>State transfer protocols</title>
                <para>
                    In order to use state transfer, a state transfer protocol has to be included in the configuration.
                    This can either be STATE_TRANSFER, STATE, or STATE_SOCK. More details on the protocols can
                    be found at <xref linkend="protlist"/>.
                </para>

                <section id="STATE_TRANSFER">
                    <title>STATE_TRANSFER</title>
                    <para>
                        The is the original state transfer protocol, which used to transfer byte[] buffers. It still does
                        that, but is internally converted to call the getState() and setState() callbacks which use
                        input and output streams.
                    </para>
                    <para>
                        Note that, because byte[] buffers are converted into input and output streams, this protocol
                        should not be used for transfer of large states.
                    </para>
                    <para>
                        For details see <xref linkend="pbcast.STATE_TRANSFER"/>.
                    </para>
                </section>

                <section id="STATE">
                    <title>STATE</title>
                    <para>
                        This is the STREAMING_STATE_TRANSFER protocol, renamed in 3.0. It sends the entire state
                        across from the provider to the requester in (configurable) chunks, so that memory consumption
                        is minimal.
                    </para>
                    <para>
                        For details see <xref linkend="pbcast.STATE"/>.
                    </para>
                </section>

                <section id="STATE_SOCK">
                    <title>STATE_SOCK</title>
                    <para>
                        Same as STREAMING_STATE_TRANSFER, but a TCP connection between provider and requester is
                        used to transfer the state.
                    </para>
                    <para>
                        For details see <xref linkend="pbcast.STATE_SOCK"/>.
                    </para>
                </section>

            </section>


        </section>





        
        <section id="Disconnecting">
            <title>Disconnecting from a channel</title>

            <para>Disconnecting from a channel is done using the following method:</para>
            <programlisting language="Java">public void disconnect();</programlisting>


            <para>
                It will have no effect if the channel is already in the disconnected or closed state. If connected, it
                will leave the cluster. This is done (transparently for a channel user) by sending
                a leave request to the current coordinator. The latter will subsequently remove the leaving node
                from the view and install a new view in all remaining members.
            </para>

            <para>
                After a successful disconnect, the channel will be in the unconnected state, and may subsequently be
                reconnected.
            </para>
        </section>

        <section>
            <title>Closing a channel</title>

            <para>
                To destroy a channel instance (destroy the associated protocol stack, and release all resources),
                method <methodname>close()</methodname> is used:
            </para>
            <programlisting language="Java">public void close();</programlisting>


            <para>
                Closing a connected channel disconnects the channel first.
            </para>

            <para>
                The close() method moves the channel to the closed state, in which no further operations are allowed
                (most throw an exception when invoked on a closed channel). In this state, a channel instance is not
                considered used any longer by an application and -- when the reference to the instance is reset --
                the channel essentially only lingers around until it is garbage collected by the Java runtime system.
            </para>
        </section>
    </section>
</chapter>
