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.

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.

The MessageListener interface below provides callbacks for message reception and for providing and setting the state:

public interface MessageListener {

    void receive(Message msg);

    void getState(OutputStream output) throws Exception;

    void setState(InputStream input) throws Exception;

}

            

Method receive() is be called whenever a message is received. The getState() and setState() methods are used to fetch and set the group state (e.g. when joining). Refer to Section 3.8.11, “Getting the group's state” for a discussion of state transfer.

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

public interface MembershipListener {

    public void viewAccepted(View new_view);

    public void suspect(Object suspected_mbr);

    public void block();

    public void unblock();

}

            

Oftentimes the only callback that needs to be implemented will be viewAccepted() which notifies the receiver that a new member has joined the group or that an existing member has left or crashed. The suspect() callback is invoked by JGroups whenever a member if suspected of having crashed, but not yet excluded ^[1].

The block() 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.

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.

The unblock() 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.

public interface Receiver extends MessageListener, MembershipListener;

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.

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.

ReceiverAdapter looks as follows:

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() {}

}

            

A ReceiverAdapter is the recommended way to implement callbacks.

public interface ChannelListener {

    void channelConnected(Channel channel);

    void channelDisconnected(Channel channel);

    void channelClosed(Channel channel);

}

            

A class implementing ChannelListener can use the Channel.addChannelListener() 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.

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(). ^[2]

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 V1:(p,q,r,s,t) split into subgroups V2:(p,q,r) and V2:(s,t), the merged view might be V3:(p,q,r,s,t). In this case the MergeView would contains a list of 2 views: V2:(p,q,r) and V2:(s,t).

A channel is created using one of its public constructors (e.g. new JChannel()).

The most frequently used constructor of JChannel looks as follows:

public JChannel(String props) throws Exception;

The props 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).

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

JChannel ch=new JChannel("/home/bela/udp.xml");

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.

For example, the Draw demo can be launched as follows:

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

or

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

In the latter case, an application downloads its protocol stack specification from a server, which allows for central administration of application properties.

A sample XML configuration looks like this (edited from udp.xml):

<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">

    <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"/>



    <PING timeout="2000"

            num_initial_members="3"/>

    <MERGE2 max_interval="30000"

            min_interval="10000"/>

    <FD_SOCK/>

    <FD_ALL/>

    <VERIFY_SUSPECT timeout="1500"  />

    <BARRIER />

    <pbcast.NAKACK use_stats_for_retransmission="false"

                   exponential_backoff="0"

                   use_mcast_xmit="true"

                   retransmit_timeout="300,600,1200"

                   discard_delivered_msgs="true"/>

    <UNICAST timeout="300,600,1200"/>

    <pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000"

                   max_bytes="4M"/>

    <pbcast.GMS print_local_addr="true" join_timeout="3000"

                view_bundling="true"/>

    <UFC max_credits="2M"

         min_threshold="0.4"/>

    <MFC max_credits="2M"

         min_threshold="0.4"/>

    <FRAG2 frag_size="60K"  />

    <pbcast.STATE_TRANSFER />

</config>

            

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

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.

Each element has to be the name of a Java class that resides in the org.jgroups.protocols package. Note that only the base name has to be given, not the fully specified class name ( UDP instead of org.jgroups.protocols.UDP). 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 com.sun.eng.protocols.reliable.UCAST.

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.

Note that all members in a group have to have the same protocol stack.

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);

        }

    }


}

                    

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 3c7e52ea-4087-1859-e0a9-77a0d2f69f29.

For example, when we have 3 channels, using logical names we might see a view "{A,B,C}", which is nicer than "{56f3f99e-2fc0-8282-9eb0-866f542ae437, ee0be4af-0b45-8ed6-3f6e-92548bfa5cde, 9241a071-10ce-a931-f675-ff2e3240e1ad} !"

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.

The logical name can be set using:

public void setName(String logical_name);

This must be done before 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.

When JGroups starts, it prints the logical name and the associated physical address(es):

-------------------------------------------------------------------
GMS: address=mac-53465, cluster=DrawGroupDemo, physical address=192.168.1.3:49932
-------------------------------------------------------------------

The logical name is mac-53465 and the physical address is 192.168.1.3:49932. The UUID is not shown here.

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 subclasses of UUID.

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.

To use custom addresses, an implementation of org.jgroups.stack.AddressGenerator has to be written.

For any class CustomAddress, it will need to get registered with the ClassConfigurator in order to marshal it correctly:

class CustomAddress extends UUID {

    static {

        ClassConfigurator.add((short)8900, CustomAddress.class);

    }

}

            

Set the address generator in JChannel: setAddressGenerator(AddressGenerator). This has to be done before the channel is connected.

An example of a subclass is org.jgroups.util.PayloadUUID, and there are 2 more shipped with JGroups.

When a client wants to join a cluster, it connects to a channel giving the name of the cluster to be joined:

public void connect(String cluster) throws Exception;

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 ^[3] ).

The connect() method returns as soon as the cluster has been joined successfully. If the channel is in the closed state (see Figure 3.2, “Channel states”), 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 coordinator. A coordinator is in charge of installing new views whenever the membership changes ^[4] .

Clients can also join a cluster and fetch cluster state in one operation. 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.

public void connect(String cluster, Address target, long timeout) throws Exception;

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.

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

public Address getAddress();

Method getClusterName() returns the name of the cluster which the member joined.

public String getClusterName();

Again, the result is undefined if the channel is in the disconnected or closed state.

The following method can be used to get the current view of a channel:

public View getView();

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

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.

Once the channel is connected, messages can be sent using one of the send() methods:

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;

            

The first send() 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).

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

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

Here's an example of sending a message to all members of a cluster:

Map data; // any serializable data

channel.send(null, data);

            

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.

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

Map data;

Address receiver=channel.getView().getMembers().get(0);

channel.send(receiver, "hello world");

            

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

<config>

    <UDP/>

    <PING />

    <FD_ALL/>

    <pbcast.NAKACK use_mcast_xmit="true"

                   discard_delivered_msgs="true"/>

    <UNICAST timeout="300,600,1200"/>

    <RSVP />

    <pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000"

                   max_bytes="4M"/>

    <pbcast.GMS print_local_addr="true" join_timeout="3000"

                view_bundling="true"/>

    ...

</config>

        

Message msg=new Message(null, null, "hello world");

msg.setFlag(Message.RSVP);

ch.send(msg);

                

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

public void receive(Message msg);

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:

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");

            

As shown above, the viewAccepted() callback of ReceiverAdapter can be used to get callbacks whenever a cluster membership change occurs. The receiver needs to be set via JChannel.setReceiver(Receiver).

As discussed in Section 3.2.4, “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.

A newly joined member may want to retrieve the state of the cluster before starting work. This is done with getState():

public void getState(Address target, long timeout) throws Exception;

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.

To participate in state transfer, both state provider and state requester have to implement the following callbacks from ReceiverAdapter (Receiver):

public void getState(OutputStream output) throws Exception;

public void setState(InputStream input) throws Exception;

            

Method getState() is invoked on the state provider (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.

The setState() method is invoked on the state requester; 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.

In a cluster consisting of A, B and C, with D joining the cluster and calling Channel.getState(), the following sequence of callbacks happens:

The following code fragment shows how a group member participates in state transfers:

public void getState(OutputStream output) throws Exception {

    synchronized(state) {

        Util.objectToStream(state, new DataOutputStream(output));

    }

}


public void setState(InputStream input) throws Exception {

    List<String> list;

    list=(List<String>)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);

    }

}

            

This code is the Chat example from the JGroups tutorial and the state here is a list of strings.

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().

Util.objectToStream(state,

                    new BufferedOutputStream(

                        new DataOutputStream(output)));

                

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.

Disconnecting from a channel is done using the following method:

public void disconnect();

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.

After a successful disconnect, the channel will be in the unconnected state, and may subsequently be reconnected.

To destroy a channel instance (destroy the associated protocol stack, and release all resources), method close() is used:

public void close();

Closing a connected channel disconnects the channel first.

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.