
[[user-channel]]
== API

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.

All of the classes discussed here are in the org.jgroups package unless otherwise mentioned.
    
[[Message]]
=== Message
Data is sent between members in the form of messages. A message can be sent by a member to
a _single member_, or to _all members_ of the group.

The structure of a message is shown in <<MessageFig>>.

[[MessageFig]]
.Structure of a message
image::./images/Message.png[Message]

A message has the following fields:

Destination address:: The address of the receiver. If `null`, the message will be sent to all
                    current group members. `Message.getDest()` returns the destination address of a message.

Source address:: The address of the sender. Can be `null`, 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.

Flags:: Flags are used to override default behavior. For example, when setting the out-of-band flag `OOB` in a message,
        the delivery order of it will get changed. For details, refer to section on <<MessageFlags,message flags>>.

Headers:: 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 `putHeader()`, `getHeader()` and `removeHeader()`
          of `Message` can be used to manipulate headers. +
          Note that headers are mainly used by protocols.

Payload:: The actual data (e.g. a byte array). The `Message` interface contains convenience method definitions to
          set and get the different types of payloads, e.g. a byte array, an object, an NIO `ByteBuffer` and so on. +
          Payloads are defined by the <<MessageTypes,different implementations>> of `Message`.



The `Message` interface is defined below (edited for legibility):

[source,java]
----
public interface Message extends SizeStreamable, Constructable<Message> {

    /** The type of the message */
    short BYTES_MSG=0, NIO_MSG=1, EMPTY_MSG=2, OBJ_MSG=3,
          COMPOSITE_MSG=5, FRAG_MSG=6, LONG_MSG=7;

    /** Returns the type of the message, e.g. BYTES_MSG, OBJ_MSG etc */
    short                 getType();

    /** Returns the destination address */
    Address               getDest();

    /** Sets the destination address */
    <T extends Message> T setDest(Address new_dest);

    /** Returns the address of the sender */
    Address               getSrc();

    /** Sets the address of the sender of this message */
    <T extends Message> T setSrc(Address new_src);

    /** Adds a header to the message */
    <T extends Message> T putHeader(short id, Header hdr);

    /** Gets a header from the message */
    <T extends Header> T  getHeader(short id);

    /** Sets one or more flags */
    <T extends Message> T setFlag(Flag... flags);

    /** Sets one or more transient flags */
    <T extends Message> T setFlag(TransientFlag... flags);

    /** Clears a number of flags */
    <T extends Message> T clearFlag(Flag... flags);

    /** Clears a number of transient flags */
    <T extends Message> T clearFlag(TransientFlag... flags);

    /** Returns true if a flag is set */
    boolean               isFlagSet(Flag flag);

    /** Returns true if a transient flag is set */
    boolean               isFlagSet(TransientFlag flag);

    /** Returns true if the message has a payload */
    boolean               hasPayload();

    /** Returns true if this message has a byte array as payload */
    boolean               hasArray();

    /**
     * Returns a reference to the payload (byte array). Throws an exception if
     * the message does not have a byte array payload (hasArray() is false).
     */
    byte[]                getArray();

    /** Returns the offset of the byte array at which user data starts.
     * Throws an exception if the message does not have a byte array payload.
     */
    int                   getOffset();

    /** Returns the length of the byte array payload. If the message does not
     * have a byte array payload, then the serialized size may be returned,
     * or an implementation may throw an exception */
    int                   getLength();

    /**
     * Sets the byte array in a message. Throws an exception if the
     * message does not have a byte array payload.
     */
    <T extends Message> T setArray(byte[] b, int offset, int length);

    /**
     * Sets the byte array in a message. Throws an exception if the message
     * does not have a byte array payload.
     */
    <T extends Message> T setArray(ByteArray buf);

    /**
     * Gets an object from the payload. If the payload is a byte array,
     * an attempt to de-serialize the array into an object is made, and
     * the object returned. If the payload is an object (e.g. as in
     * an ObjectMessage), the object will be returned directly.
     */
    <T extends Object> T  getObject();

    /**
     * Sets an object in a message. In a ObjectMessage, the object is
     * set directly.
     * In a BytesMessage, the object is serialized into a byte array
     * and set as the payload of the message.
     */
    <T extends Message> T setObject(Object obj);

    /**
     * Returns the payload without any conversion (e.g. as in {@link #getObject()} or {@link #getArray()})
     */
     <T extends Object> T getPayload();

     /**
      * Sets the payload
      * @param pl The paylolad
      */
     Message              setPayload(Object pl);


    /** Returns the exact size of the marshalled message */
    int                   size();

    enum Flag {
        OOB,            // message is out-of-band
        DONT_BUNDLE,    // don't bundle message at the transport
        NO_FC,          // bypass flow control
        NO_RELIABILITY, // bypass UNICAST3 and NAKACK2
        NO_TOTAL_ORDER, // bypass total order (e.g. SEQUENCER)
        NO_RELAY,       // bypass relaying (RELAY2)
        RSVP,           // ack of a multicast
        RSVP_NB,        // non blocking RSVP
        INTERNAL,       // for internal use by JGroups only, don't use!
        SKIP_BARRIER;   // passing messages through a closed BARRIER
    }

    enum TransientFlag {
        OOB_DELIVERED,
        DONT_LOOPBACK;   // don't loopback if set and a multicast message
    }
}
----

`Message` defines methods to get and set the destination and sender's address, set/clear flags and add and remove
headers. These methods are implemented in `BaseMessage`, which is extended by all message implementations.

The rest of the methods are defined to get and set the payload. They're all generic, and implementations may or may
not choose to implement them.

The table below describes the payload-related methods of `Message`:

.Payload related methods
[width="100%",cols="1,10",frame="topbot",options="header"]
|==========================
|Name    |Description
|hasPayload | Returns true if the message has a payload, e.g. a byte array or an object.
              This is more generic than `hasArray()`, as it is not just applicable to a byte array.
|hasArray | Returns true if the payload is a byte array (even if it's null).
|getArray | Returns a reference to the byte array (if the payload is a byte array). May throw an exception
            if the payload is not a byte array (`hasArray()` is false).
|getOffset | Returns the offset of the payload in the byte array.
|getLength | Returns the length of the payload.
|setArray  | Sets the byte array in a message. Throws an exception if the
             message does not have a byte array payload (hasArray() is false).
|getObject | Gets an object from the payload. If the payload is a byte array, an attempt to de-serialize
             the array into an object is made, and the object returned. +
             If the payload is an object (e.g. as in an `ObjectMessage`), the object will be returned directly.
|setObject | Sets an object in a message. In a `ObjectMessage` (see below), the object is set directly. In a
             `BytesMessage`, the object is serialized into a byte array and set as the payload of the message.
|getPayload | Gets the payload from a message. The actual payload without any conversions is returned. E.g. compared
to `getObject()`, in a `BytesMessage` the actual byte array would be returned with `getPayload()`, whereas
`getObject()` would try to deserialize the byte array into an object.
|setPayload | Sets the payload
|==========================


[[MessageTypes]]
=== Message types

The `Message` interface has a number of subclasses, each having a different payload. <<BytesMessage,`BytesMessage`>>
for example can be used with byte arrays, <<NioMessage,`NioMessage`>> handles `ByteBuffer` types and
<<ObjectMessage,`ObjectMessage`>> accepts (serializable) `Objects` as payload.

A second set of message types operates on other messages and does not provide a payload of its own:

<<FragmentedMessage,`FragmentedMessage`>> wraps any other message that's too big to be sent by the transport (e.g.
UDP has a datagram size limit of ~65K) with a fragmentation size, and creates fragments of the underlying message
when sent. On the receiver side, the fragments are cached and the original message is re-created when all fragments
have been received and then sent up.

<<CompositeMessage,`CompositeMessage`>> wraps multiple messages and marshals them at send time. For example, if the
application has an NIO `ByteBuffer` and a 4-byte command (byte array), then rather than creating a byte array, copying
the command and the `ByteBuffer` into it, and passing the resulting byte array to the message in 4.x, in 5.0 the
application can simply create a `CompositeMessage` and pass the command and `ByteBuffer` to it, and the
`CompositeMessage` will marshal both at send time.

The available message types are discussed below.


==== Late marshalling
Contrary to JGroups 4.x, the `Message` subclasses perform *late marshalling*: rather than having to marshal an object
into a byte array (to be passed to a 4.x `Message`), the object in the payload of an `ObjectMessage` is serialized
at send time only, possibly directly into a network socket's buffer (or output stream).

The following example shows 4.x code:

[source,java]
----
MyObject obj=...;
byte[] buffer=Util.objectToByteBuffer(obj); // memory allocation
Message msg=new Message(null, buffer);
channel.send(msg);
----

The object is serialized into a byte array, which is passed to the `Message` constructor. At the network level, that
byte array will get written to the output stream of the socket. The `buffer` byte array is a temporary copy, and
introduces additional memory allocation.

In 5.0, we can eliminate the allocation:

[source,java]
----
MyObject obj=...;
Message msg=new ObjectMessage(null, obj);
channel.send(msg);
----

The `ObjectMessage` type does not serialize the object until the message itself is serialized (at send time). The
entire message including the object payload is written directly to the socket's output stream.
The elimination of memory allocation for those temporary byte arrays reduces overall memory pressure, possibly leading
to better performance.


[[MessageFactory]]
==== MessageFactory

JGroups 5.0 comes with a number of message types (see the next sections). If none of them are a fit for the application's
requirements, new message types can be defined and registered. To do this, the new message type needs to implement
`Message` (typically by subclassing `BaseMessage`) and registering it with the `MessageFactory` in the transport:

[source,java]
----
CustomMessage msg=new CustomMessage(...);
JChannel ch;
TP transport=ch.getProtocolStack().getTransport();
MessageFactory mf=transport.getMessageFactory();
mf.register((short)12345, CustomMessage::new)
----

A (unique) ID has to be assigned with the message type, and then it has to be registered with the message factory
in the transport. This has to be done before sending an instance of the new message type.
If the ID has already been registered before, or is taken, an exception will be thrown.
Note that the default implementation of `MessageFactory` requires all IDs to be greater than 32, so that there's room
for adding built-in message types.

NOTE: It is recommended to register all custom message types _before_ connecting the channel, so that potential errors
are detected early.

[[CustomMessageFactory]]
==== Custom `MessageFactory`
`MessageFactory` is a simple interface:

[source,java]
----
public interface MessageFactory {
    <T extends Message> T create(short id);
    void register(short type, Supplier<? extends Message> generator);
}
----
We saw the that the `register()` method is used to associate new message types with IDs <<MessageFactory,above>>.

There is a `DefaultMessageFactory` which is set in the transport (`TP`). If more control over the creation of custom
messages is desired, a custom implementation of `MessageFactory` can be written and registered in the transport, using
`TP.setMessageFactory(MessageFactory mf)`.

An example for why we might want to provide our own `MessageFactory` is that we have control over the creation of
messages; e.g. to create an `NioMessage` with a *direct* `ByteBuffer`, we may want to use a _pool_ of off-heap memory
rather than calling `ByteBuffer.allocateDirect()` for each message, which is slow.


[[BytesMessage]]
==== BytesMessage
This is the equivalent to the 4.x `Message`, and contains a byte array, offset and length. There are methods to get and
set the byte array from a byte array, NIO `ByteBuffer`, Object etc. The latter case marshals the object into a byte
array and sets it in the message. Conversely, `getObject()` tries to unmarshal the byte array into an object.

NOTE: It is recommended to only use the methods which get and set a *byte array*, as the other methods may get
deprecated in the future. See the section on <<PayloadMismatch,payload mismatch>> below for details.

The simplest way to convert 4.x applications to 5.0 is:

[source,java]
----
Message msg=new Message(null, "hello world".getBytes()); // 4.x
----

[source,java]
----
Message msg=new BytesMessage(null, "hello world".getBytes()); // 5.0
----



[[NioMessage]]
==== NioMessage
An `NioMessage` has a `ByteBuffer` as payload. The `ByteBuffer` can be heap-based or direct (off-heap). A heap-based
buffer will be created with heap memory again when received. For an off-heap (direct) buffer, we can choose whether
heap memory should be used for the buffer when receiving an `NioMessage`, or whether off-heap (direct) memory should
be used. See the `useDirectMemory()` method below.

Alternatively, a custom message factory could manage a pool of off-heap memory and create the buffer in the
`NioMessage` with memory from that pool.

The methods of `NioMessage` are:

[width="100%",cols="2,6",frame="topbot",options="header"]
|==========================
|Name    |Description
|isDirect() | Returns true if the buffer is off-heap, false otherwise
|getBuf() | Returns the `ByteBuffer` payload
|setBuf(ByteBuffer) | Sets the `ByteBuffer` payload
|useDirectMemory() | When true, the `ByteBuffer` will be created on-heap when unmarshalled, otherwise
                             it will be created off-heap (direct `ByteBuffer`). See the setter below for more details. +
                             Default: `false`
|useDirectMemory(boolean b) | If true, use direct memory when creating `ByteBuffer` payloads,
                         e.g. on `setArray(byte a)`, `setArray(ByteArray)`, `setObject(Object)` or when getting read
                         from the network (`readPayload(DataInput)`). +
      Unless this flag is set to true, a direct `ByteBuffer` payload will become a heap-based payload by a
      receiver when received from the network. This may be useful if we want to use off-heap (direct) memory only for
      sending, but not receiving of messages. +
      If we want the receiver to create a direct-memory based message, set this flag to true before sending the message.
|==========================

NOTE: The envisioned use case for `useDirectMemory()` is when we send an `NioMessage` with a direct `ByteBuffer`, but
      don't need the `ByteBuffer` to be created in off-heap memory at the receiver, when on-heap will do. +
      The alternative is to provide a custom <<MessageFactory,`MessageFactory`>>.



[[EmptyMessage]]
==== EmptyMessage
As its name implies, an `EmptyMessage` carries no payload. This means that it uses less memory, e.g. compared to
`BytesMessage`, it has 3 fields (array, offset, length) less, and so has ~12 bytes less.

JGroups itself uses quite some messages that send around no payload, only headers, and `EmptyMessage` instances are
ideal for this.

Here's an example that sends a heartbeat in `FD_ALL`:

[source,java]
----
Message heartbeat=new EmptyMessage()
                   .setFlag(Message.Flag.INTERNAL)
                   .putHeader(id, new HeartbeatHeader());
----

While this message type is mainly used internally, it _is_ a public class and can as such be used by applications, too.




[[ObjectMessage]]
==== ObjectMessage
An `ObjectMessage` has a Plain Old Java Object (POJO) as payload. The object payload is marshalled directly into the
output stream _at send time_, using `writeTo()`.

If the object implements `SizeStreamable`, then the `writeTo()` and `readFrom()` methods will be used for efficient,
application-controlled, marshalling. Otherwise, for some object types ('primitive types such as `int`, `Boolean`,
`byte[]` etc), JGroups will try to do the marshalling.

For all other cases, the the object is wrapped into an `ObjectWrapper` which uses just-in-time serialization to generate
a byte array, which is later written to the output stream.

WARNING: When the byte array is set, the contents of the object should not be changed anymore, or else the old state
         of the object will be sent on serialization. In case the object needs to be changed nevertheless, then
         method `setObject(Object)` can be used: it will set the object and null the serialized format byte array, so
         at marshalling time, it will be set again.

To get and set the payload, methods `getObject()` and `setObject()` should be used. Methods `getArray()` /
`setArray()` will throw an exception.

Sample code:

[source,java]
----
Message msg=new ObjectMessage(null, obj);           // constructor
Message msg=new ObjectMessage(null).setObject(obj); // setter
MyObject obj=msg.getObject();
----




[[LongMessage]]
==== LongMessage
A `LongMessage` has a `long` as payload. This allows for simple sending of `longs` and `integers`, without having
to marshal them into byte arrays.

These types are used for example in the flow control protocols to send credit-requests and responses.



[[CompositeMessage]]
==== CompositeMessage
This message type has multiple messages as payload. The messages have to have the same destination.

It solves the following problem (4.x code):

[source,java]
----
final String hello="hello world";
byte[] metadata=createMetadata();                 // metadata
ByteBuffer cmd=ByteBuffer.wrap(hello.getBytes()); // the command

byte[] arr=new byte[metadata.length + cmd.remaining()];  // temp memory allocation!
System.arraycopy(metadata, 0, arr, 0, metadata.length);
System.arraycopy(cmd.array(), cmd.arrayOffset(), arr, metadata.length, cmd.remaining());
Message msg=new Message(null, arr);
// send the message consisting of metadata and command

public void receive(Message msg) {
    byte[] array=msg.getRawBuffer(), metadata=new byte[4];
    int offset=msg.getOffset(), len=msg.getLength();

    System.arraycopy(array, offset, metadata, 0, metadata.length);
    ByteBuffer cmd=ByteBuffer.allocate(hello.length);
    cmd.put(array, metadata.length, len-metadata.length);
    // process metadata and command
}
----

In the above code, we have to create a new byte array and copy the metadata and command into it. Then a message with the
combined data is sent. At the receiver, we have to divide the combined byte array into the metadata and command
portions again.

This code can be rewritten using `CompositeMessage` as follows (5.0 code):

[source,java]
----
final String hello="hello world";

byte[] metadata=createMetadata();
ByteBuffer cmd=ByteBuffer.wrap(hello.getBytes());

Message msg=new CompositeMessage(null, new BytesMessage(null, metadata),
                                       new NioMessage(null, cmd));
// send the message with metadata and command

public void receive(Message msg) {
    CompositeMessage cm=(CompositeMessage)msg;
    BytesMessage m1=cm.get(0);
    NioMessage m2=cm.get(1);
    // process metadata (m1) and command (m2)
}
----

Here, we don't have to combine the two pieces of data into one, but simply pass them to a `CompositeMessage`, which
is then sent. At the receiver, the received `Message` is narrowed to a `CompositeMessage` and the individual messages
can simply be accessed by index.

This eliminates 1 memory allocation, and it also simplifies programming a bit.

The methods of `CompositeMessage` are:

[width="80%",cols="2,4",frame="topbot",options="header"]
|==========================
|Name    |Description
|getNumberOfMessages() | Returns the number of messages
|add(Message) | Adds a message, increasing the capacity if needed
|add(Message ... msgs) | Adds a number of messages
|get(int index) | Returns the message at the given index
|collapse(boolean)|Collapses the payload and sends the message as a `BytesMessage` (see next section)
|==========================

For a detailed listing of all methods consult the javadoc.

===== Collapsing a CompositeMessage
When we send a `CompositeMessage` consisting of a `BytesMessage` of 1000 bytes and an `NioMessage` of 50 bytes,
we may want to receive a `BytesMessage` of 1050 bytes, with the payloads of the individual messages
*collapsed*, rather than a `CompositeMessage`.

This is oftentimes the case when we would like to combine the 2 payloads at the sender side into 1, but
want to avoid the overhead of the byte array creation of size 1050, but don't mind handling the full 1050-byte
array at the receiver side.

The code below shows how to use `collapse()`:

[source,java]
----
public void testCollapse2() throws Exception {
    CompositeMessage msg=new CompositeMessage(DEST)
        .add(new BytesMessage(DEST, "hello".getBytes()))
        .add(new NioMessage(DEST, ByteBuffer.wrap(" world".getBytes())))
        .add(new ObjectMessage(DEST, "hello"))
        .collapse(true);
    int length=msg.getLength();
    ByteArray buf=Util.messageToBuffer(msg);
    Message msg2=Util.messageFromBuffer(buf.getArray(),
                                        buf.getOffset(),
                                        buf.getLength(), MF);
    assert msg2 instanceof BytesMessage;
    assert msg2.getLength() == length;

    byte[] bytes=msg2.getArray();
    String s=new String(bytes, 0, 11);
    assert s.equals("hello world");
    DataInput in=new ByteArrayDataInputStream(bytes,
                                              s.length(),
                                              bytes.length-s.length());
    ObjectMessage om=new ObjectMessage();
    om.readPayload(in);
    assert om.getObject() instanceof String;
    assert om.getObject().equals("hello");
}
----

Here, we're adding a byte array, an NIO `ByteBuffer` and an object and then set `collapse()` in the
`CompositeMessage`.

This means that - at serialization time - the byte array is written first (without the size), then the
`ByteBuffer`, then the object.

When receiving the message, we see that it's a `BytesMessage`, _not_ a `CompositeMessage`. We know that the
payloads of the `BytesMessage` (`"hello"`) and the payload of the `NioMessage` (`" world"`) have been combined
and can therefore read the entire 11 bytes directly into a string `"hello world"`.

After this, we read the object (via an `ObjectMessage`).

NOTE: Payload collapsing is envisaged to be used mainly for message types which have an array (e.g. `BytesMessage`
or `NioMessage`).



[[FragmentedMessage]]
==== FragmentedMessage
A `FragmentedMessage` wraps another message (also a `CompositeMessage`!), with an offset and length. The offset/length
combo defines a subrange of the underlying message that is to be marshalled.

For example, if we have an `ObjectMessage` whose length is 500, and the fragmentation size is 200, then 3
`FragmentedMessage` instances will be created, with ranges `0..199`, `200..399` and `400..499`.

The second `FragmentedMessage` will only marshall the part of the underlying message between indices `200` and `399`.

`FragmentedMessage` is only used internally by the fragmentation protocols, and not by users. It is discussed here
mainly for completeness.


[[Refcounting]]
==== Reference-counted messages
When a sender wants to reuse the message payload, a shared pool of payloads may be used. For example, we may want to
create a pool of direct ByteBuffers (allocated off-heap) and wrap ByteBuffers (taken from that pool) in NioMessages.

After sending an NioMessage with ByteBuffer `buf`, we may want to modify `buf` and send another NioMessage with `buf`
as payload. However, this might fail, as the message may not yet have been sent, or it may have been dropped and
will later get retransmitted.

For example, if we set `buf` to `123`, send it, then set it to `456` and send it again, the following
might happen:

* `123` is sent as NioMessage `m1`, but `m1` is dropped
* `456` is sent as NioMessage `m2`
* `m1` gets retransmitted. By now, its payload is `456`, as `buf` has been changed in the meantime
* This leads to a receiver receiving `456` twice, rather than `123` followed by `456`.

Reference counting makes sure that a payload is only available to be reused when

* a unicast message has been sent to a receiver and the ack from the receiver has been received by the sender,
so that the message can be purged from the retransmission table, or
* a multicast message (= message to all members) has been seen by all members and has been purged from
the retransmission table.

This is achieved by reference counting: a message increments a reference count (refcnt) when being sent,
and decrements it when `JChannel.send()` returns. Retransmission protocols (such as `NAKACK2` or `UNICAST3`)
increment refcnt, and decrement it when the message is known to have been received by the receiver(s).

When refcnt becomes 0, a callback is invoked (with the message as argument), which can (e.g.) return the
message to a pool.

As example of a refcounted message is `RefcountedNioMessage`:

[source,java]
----
public class RefcountedNioMessage extends NioMessage implements Refcountable<Message> {
    protected final RefcountImpl<Message> impl=new RefcountImpl<>();

    public RefcountedNioMessage() {
    }

    public RefcountedNioMessage(Address dest) {
        super(dest);
    }

    public RefcountedNioMessage(Address dest, ByteBuffer buf) {
        super(dest, buf);
    }

    public synchronized byte getRefcount() {
        return impl.getRefcount();
    }

    @Override public synchronized RefcountedNioMessage incr() {
        impl.incr();
        return this;
    }

    @Override public synchronized RefcountedNioMessage decr() {
        impl.decr(this);
        return this;
    }

    public RefcountedNioMessage onRelease(Consumer<Message> rc) {
        impl.onRelease(rc);
        return this;
    }
}
----

It subclasses `NioMessage` and implements methods `incr()` and `decr()` of the `Refcountable` interface. The latter calls
a function previously passed in via `onRelease()`. This function may for example return the payload of the NioMessage
to a pool. See `RefcountedNioMessageTest` as an example.

NOTE: Reference counting is an experimental feature, introduced in 5.1.0, and is WIP. The authors are interested in
scenarios where this would be beneficial, and its performance impact. Please contact us if you are serious
about using this feature; we're interested in feedback, suggestions and are happy to collaborate on this!

===== Reference counting on the receiver side
Reference counting is currently only implemented on the sender side.

However, it could also be useful on the receiver side: a `MessageFactory` could create instances of ref-counted messages
(the payload would be taken from a buffer pool) and increment the refcount.

When such a message is received and processed by the application, the `receive(Message)` callback (or, alternatively,
the application itself) could decrement the refcount, thus returning the buffer to the pool.



[[PayloadMismatch]]
=== Mismatch between message type and getters/setters

The `Message` interface has a number of generic methods to get/set payloads of various types.

It makes sense to call `getArray()` / `setArray()` on a `BytesMessage` and `getObject()` / `setObject()` on an
`ObjectMessage`.

However, there are mismatches between message type and getters / setters:

* `BytesMessage` and `getObject()`: the byte array payload will need to be de-serialized into an object. This is slower
than using an `ObjectMessage` instead and calling `getObject()` on it to get the object. +
If an object needs to be serialized into a byte array and we want to use a `BytesMessage`, then it is better if the
_application_ serializes the object and passes the byte array to the `BytesMessage`, rather than JGroups doing
the serialization: the application has more knowledge of its types and can therefore do a much better job at
serialization than the generic algorithm used by JGroups.
* `CompositeMessage` and `getArray()` / `getObject()`: this will throw an exception
* `NioMessage` and `getArray()`: this works, but will create a new byte array (unneeded memory allocation)

In these cases, it makes sense to narrow the `Message` to its actual type and use the subclass:

[source,java]
----
public void receive(Message msg) {
    NioMessage m=(NioMessage)msg;
    ByteBuffer buf=m.getBuf();
    // do something directly with the ByteBuffer
}
----

After all, the application knows what message types it sends, and can therefore safely downcast.


[[Receiver]]
=== Receiver

The `Receiver` interface defines the callbacks that are invoked at the receiver side:

[source,java]
----
public interface Receiver {

    default void receive(Message msg) {
    }

    default void receive(MessageBatch batch) {
        for(Message msg: batch) {
            try {
                receive(msg);
            }
            catch(Throwable t) {
            }
        }
    }

    default void viewAccepted(View new_view) {}

    default void block() {}

    default void unblock() {}

    default void getState(OutputStream output) throws Exception {
        throw new UnsupportedOperationException();
    }

    default void setState(InputStream input) throws Exception {
        throw new UnsupportedOperationException();
    }
}
----

[width="90%",cols="2,5",frame="topbot",options="header"]
|==========================
|Name    |Description
|receive(Message)| A message is recived
|receive(MessageBatch)| A <<MessageBatch,message batch>> is received
|viewAccepted|Called when a change in membership has occurred. No long running actions, sending of messages
              or anything that could block should be done in this callback. If some long running action
               needs to be performed, it should be done in a separate thread.
|block|Called (usually by the <<FLUSH,FLUSH>> protocol), as an indication that the member should stop sending
       messages. Any messages sent after returning from this callback might get blocked by the FLUSH protocol.
       When the FLUSH protocol is done, and messages can be sent again, the FLUSH protocol will simply unblock
       all pending messages. If a callback for unblocking is desired, implement `unblock()`.
|unblock|Called _after_ the FLUSH protocol has unblocked previously blocked senders, and messages can be sent again.
|getState|Allows an application to write the state to an OutputStream. After the state has been written, the
          OutputStream doesn't need to be closed as stream closing is automatically done when a calling thread
          returns from this callback.
|setState|Allows an application to read the state from an InputStream. After the state has been read, the InputStream
          doesn't need to be closed as stream closing is automatically done when a calling thread
          returns from this callback. +
          Refer to <<StateTransfer,State transfer>> for a discussion of state transfer.
|==========================


CAUTION: Note that anything that could block should _not_ 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! +
If we need to send a message in a callback, the sending should be done on a separate thread, or a
task should be submitted to the timer.

A `Receiver` callback is registered with the channels via `JChannel.setReceiver(Receiver)`.




[[ChannelListener]]
==== ChannelListener

[source,java]
----
public interface ChannelListener {
    void channelConnected(JChannel channel);
    void channelDisconnected(JChannel channel);
    void channelClosed(JChannel channel);
}
----

A class implementing ChannelListener can use the JChannel.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.
            

[[Address]]
=== Address

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:
        
[source,java]
----
public interface Address extends Streamable, Comparable<Address> {
    int size();
}
----

For marshalling purposes, `size()` needs to return the number of bytes an instance of an address implementation
takes up in serialized form.
        
WARNING: Please never use implementations of Address directly; Address should always be used as an opaque identifier of
a cluster node!
        
Actual implementations of addresses are generated by the transport protocol (e.g. `UDP` or `TCP`).
This allows for all possible types of addresses to be used with JGroups.
        
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).
        
The default implementation of Address is `org.jgroups.util.UUID`. It uniquely identifies
a node, and when disconnecting and reconnecting to a cluster, a node is given a new UUID on reconnection.
        
UUIDs are never shown directly, but are usually shown as a logical name (see <<LogicalName,Logical names>>).
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.
        
UUIDs maps to IpAddresses, which are IP addresses and ports. These are eventually used by the transport
protocol to send a message.




[[MessageBatch]]
=== MessageBatch

A message batch is a class used to deliver messages which includes a number of messages rather than just one. The sender
and destination (= receiver) of a batch _is the same for all messages of the batch_. A batch can be iterated over, e.g.

[source.java]
----
MessageBatch batch;
for(Message msg: batch) {
    // do something with msg
}
----

The advantage of a message batch is that multiple messages are delivered in one go; which means potential locks are
acquired only once, we have fewer threads (less work for the thread pool) and fewer context switches.

JGroups tries to bundle as many messages as possible into a batch on the sender side.

Also on the receiver side, if multiple threads added messages to a table, it tries to remove as many of them as possible
and pass them up to other protocols (or the application) as a batch.


[[Header]]
=== Header

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.
        

[[Event]]
=== Event

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.
        
NOTE: Headers and events are only used by protocol _implementers_; they are not needed by application code!
            

[[View]]
=== View

A view (`org.jgroups.View`) is a list of the current members of a group. It consists
of a `ViewId`, 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.
        
Note that the first member of a view is the _coordinator_ (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.
        
The code below shows how to send a (unicast) message to the first member of a view (error checking code
omitted):
        
[source,java]
----
View view=channel.getView();
Address first=view.getMembers().get(0);
Message msg=new ObjectMessage(first, "Hello world");
channel.send(msg);
----

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

[[ViewId]]
==== ViewId

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

NOTE: Note that the latter 2 methods only take the ID into account.
            

[[MergeView]]
==== MergeView

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. MergeView is
a subclass of View and contains as additional instance variables the list of views that were merged.

As an example if the cluster with 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 contains a list of
two views: V2={P,Q,R}) and V2={S,T}.

NOTE: Because the default merge policy adds members from subgroups into a common group and sorts the resulting list, the
membership order might change on a merge event. Thus a view V1={P,Q,R,S,T}, followed by view V2={P,Q,R} and V2={S,T}
might result in a merge view V3={P,T,Q,S,R}. To prevent this, the task of creating new views can be delegated to
custom code (see <<MembershipChangePolicy>>).

WARNING: Because merging needs to handle all edge cases, it is *not* guaranteed that subsequent MergeViews won't have
identical membership. For example, we we have view A2={A,B} in A and B3={B} in B, then a subsequent merge might install
view A4={A,B} in both A and B. In A's case, the membership between A2 and A4 doesn't change. An application has to be
able to handle duplicate subsequent merge views. Note that consecutive _regular_ views will never have duplicate members.


[[JChannel]]
=== JChannel

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.
        

A state transition diagram for the major states a channel can assume are shown in <<ChannelStatesFig>>.
        
[[ChannelStatesFig]]
image::./images/ChannelStates.png[Channel states]

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.
        
The methods available for creating and manipulating channels are discussed now.
        

[[CreatingAChannel]]
==== Creating a channel

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

The most frequently used constructor of JChannel looks as follows:
            
[source,java]
----
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:

[source,java]
----
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`):
            

[source,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.xsd">
    <UDP
         mcast_port="${jgroups.udp.mcast_port:45588}"
         ip_ttl="4"
         max_bundle_size="64K"
         enable_diagnostics="true"
         thread_pool.min_threads="2"
         thread_pool.max_threads="8"
         thread_pool.keep_alive_time="5000" />

    <PING />
    <MERGE3 max_interval="30000" min_interval="10000"/>
    <FD_SOCK/>
    <FD_ALL/>
    <VERIFY_SUSPECT timeout="1500"  />
    <pbcast.NAKACK2 xmit_interval="500" />
    <UNICAST3 xmit_interval="500" />
    <pbcast.STABLE desired_avg_gossip="50000"
                   max_bytes="4M"/>
    <pbcast.GMS print_local_addr="true" join_timeout="2000"/>
    <UFC max_credits="2M"
         min_threshold="0.4"/>
    <MFC max_credits="2M"
         min_threshold="0.4"/>
    <FRAG2 frag_size="60K"  />
</config>
----

A stack is wrapped by `<config>` and `</config>` elements and lists all protocols from bottom
(`UDP`) to top (`FRAG2`). 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: Note that all members in a group have to have the same protocol stack.
            

[[ProgrammaticCreation]]
===== Programmatic creation

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.
                
An example of how to programmatically create a channel is shown below (copied from `ProgrammaticChat`):
                
[source,java]
----
public class ProgrammaticChat {

    public static void main(String[] args) throws Exception {
        Protocol[] prot_stack={
          new UDP().setValue("bind_addr", InetAddress.getByName("127.0.0.1")), // <2>
          new PING(),
          new MERGE3(),
          new FD_SOCK(),
          new FD_ALL(),
          new VERIFY_SUSPECT(),
          new BARRIER(),
          new NAKACK2(),
          new UNICAST3(),
          new STABLE(),
          new GMS(),
          new UFC(),
          new MFC(),
          new FRAG2()};
        JChannel ch=new JChannel(prot_stack).name(args[0]); // <1>

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

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

        ch.connect("ChatCluster");
        for(;;) {
            String line=Util.readStringFromStdin(": ");
            ch.send(null, line); // causes an ObjectMessage to be created
        }
    }

}
----

First, the JChannel is created (1) with an array of protocols. The protocols have some fields already set, e.g.
`bind_addr` in `UDP` (2).

The protocols are arranged bottom-first; e.g. `UDP` as transport is 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.



[[LogicalName]]
==== Giving the channel a logical name

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:

[source,java]
----
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):
                
[listing]
....
-------------------------------------------------------------------
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.
            

[[CustomAddresses]]
==== Generating custom addresses

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:
            
[source,java]
----
class CustomAddress extends UUID {
    static {
        ClassConfigurator.add((short)8900, CustomAddress.class);
    }
}
----


NOTE: Note that the ID should be chosen such that it doesn't collide with any IDs defined in
      `jg-magic-map.xml`.

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 two more shipped with JGroups.
            

[[JoiningACluster]]
==== Joining a cluster

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


[source,java]
----
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).

NOTE: Local delivery can be turned off using `setDiscardOwnMessages(true)`.
            

The `connect()` method returns as soon as the cluster has been joined successfully. If the channel is in
the closed state (see <<ChannelStatesFig, 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


[[AtomicJoinAndGetState]]
==== Joining a cluster and getting the state in one operation

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.
            
[source,java]
----
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.
            

[[GettingAddress]]
==== Getting the local address and the cluster name

Method `getAddress()` returns the address of the channel. The address may or may
not be available when a channel is in the unconnected state.
            
[source,java]
----
public Address getAddress();
----

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

[source,java]
----
public String getClusterName();
----

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

[[GettingTheView]]
==== Getting the current view

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


[source,java]
----
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.
            

[[SendingMessages]]
==== Sending messages

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


[source,java]
----
public void send(Message msg) throws Exception;
public void send(Address dst, Object obj) 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 an object, 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:
            
[source,java]
----
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.setArray()` or `Message.setObject()`) also works.
            
Here's an example of sending a unicast message to the first member (coordinator) of a group:
            
[source,java]
----
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.


===== A note about buffer reuse
The following code is wrong:

[source,java]
----
protected void sendFile() throws Exception {
    FileInputStream in=new FileInputStream(filename);
    byte[] buf=new byte[8096];
    for(;;) {
        int bytes=in.read(buf);
        if(bytes == -1)
            break;
        channel.send(new BytesMessage(null, buf, 0, bytes));
    }
}
----

* Buffer `buf` is reused and can get overwritten with new data while JGroups
** queues the message in a bundler and sends multiple messages as a message batch
** possibly retransmits the message if not received by the receiver(s); retransmitting the changed buffer
* Correct: move `buf` into the for loop


[[DiscardingOwnMessages]]
===== Discarding one's own messages

Sometimes, it is desirable not to have to deal with one's own messages, ie. messages sent by oneself.
To do this, `JChannel.setDiscardOwnMessages(boolean flag)` 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.
                


[[RsvpSection]]
===== Synchronous messages

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

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 _before_ 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.

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.

To use RSVP, two things have to be done:

First, the `RSVP` protocol has to be in the config, somewhere above the reliable transmission
protocols such as `NAKACK2` or `UNICAST3`, e.g.:
                
[source,xml]
----
<config>
    <UDP/>
    <PING />
    <FD_ALL/>
    <pbcast.NAKACK2 use_mcast_xmit="true"
                   discard_delivered_msgs="true"/>
    <UNICAST3 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"/>
    ...
</config>
----

Secondly, the message we want to get ack'ed must be marked as `RSVP`:
                
[source,java]
----
Message msg=new ObjectMessage(null, "hello world").setFlag(Message.RSVP);
ch.send(msg);
----

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.
                
If the +timeout+ property is 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.
                
The configuration of RSVP is described here: <<RSVP>>.
                
NOTE: RSVP was added in version 3.1.

====== Non blocking RSVP

Sometimes a sender wants a given message to be resent until it has been received, or a timeout occurs, but doesn't want
to block. As an example, `RpcDispatcher.callRemoteMethodsWithFuture()` needs to return immediately, even if the results
aren't available yet. If the call options contain flag `RSVP`, then the future would only be returned once all
responses have been received. This is clearly undesirable behavior.

To solve this, flag `RSVP_NB` (non-blocking) can be used. This has the same behavior as `RSVP`, but the caller is not
blocked by the RSVP protocol. When a timeout occurs, a warning message will be logged, but since the caller doesn't
block, the call won't throw an exception.


[[ReceivingMessages]]
==== Receiving messages

Methods `receive(Message)` and `receive(MessageBatch)` in ReceiverAdapter (or Receiver) can be overridden to
receive messages.

[source,java]
----
public void receive(Message msg);
public void receive(MessageBatch batch);
----

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:

[source,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");
----

NOTE: The semantics of `receive(Message msg)` changed slightly in 4.0: as the buffer of `msg` might get reused by
the transport (to reduce the memory allocation rate), the `receive()` method must consume the buffer
(e.g. de-serialize it into an application object), or make a copy.
As soon as `receive()` returns, the message's buffer might get overwritten with new data.


To receive message batches (see <<MessageBatch>>), method `receive(MessageBatch)` has to be implemented, e.g.:

[source,java]
----
public void receive(MessageBatch batch) {
    for(Message msg: batch) {
        // do something with the message
    }
}
----

Implementing the `receive(MessageBatch)` callback is not strictly necessary, as the default implementation will call
`receive(Message)` for each message of a batch, but it may be more efficient if the application can process batches
of messages in one go.


[[ReceivingViewChanges]]
==== Receiving view changes

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

[[StateTransfer]]
==== Getting the group's state

A newly joined member may want to retrieve the state of the cluster before starting work. This is done
with `getState()`:
            
[source,java]
----
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.
            
NOTE: The reason for not directly returning the state as a result of
      getState() 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!

To participate in state transfer, both state provider and state requester have to implement the
following callbacks from ReceiverAdapter (Receiver):
            
[source,java]
----
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:
                
* D calls `JChannel.getState()`. The state will be retrieved from the oldest member, A
* A's `getState()` callback is called. A writes its state to the output stream passed as a
  parameter to `getState()`.
* 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.
* D: `JChannel.getState()` returns. Note that this will only happen _after_ 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.
                    
The following code fragment shows how a group member participates in state transfers:

[source,java]
----
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()`.

.Performance when writing to an output stream
****
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. +
[source,java]
----
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.
            

[[StateTransferProtocols]]
===== State transfer protocols

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 in the <<protlist,protocols list>> section.
                

[[STATE_TRANSFER]]
.STATE_TRANSFER

This 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.

Note that, because `byte[]` buffers are converted into input and output streams, this protocol
should not be used for transfer of large states.
                    
For details see <<pbcast.STATE_TRANSFER>>.
                    

[[STATE]]
.STATE

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.
                    
For details see <<pbcast.STATE>>.
                    

[[STATE_SOCK]]
.STATE_SOCK

Same as `STREAMING_STATE_TRANSFER`, but a TCP connection between provider and requester is
used to transfer the state.
                    
For details see <<pbcast.STATE_SOCK>>.
                    

[[Disconnecting]]
==== Disconnecting from a channel

Disconnecting from a channel is done using the following method:

[source,java]
----
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.
            

==== Closing a channel

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


[source,java]
----
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.
            

