Message.java
/*
** Module : Message.java
** Abstract : container for the most elemental protocol transmissions
**
** Copyright (c) 2005-2024, Golden Code Development Corporation.
**
** -#- -I- --Date-- --JPRM-- ----------------------------Description------------------------------
** 001 SIY 20050113 @19497 Created initial version.
** 002 SIY 20050211 @19772 Added replacement for the standard
** object reading/writing methods and copying
** of routing keys.
** 003 SIY 20050315 @20382 Organized imports, fixed spelling and other
** minor cleanups.
** 004 GES 20060726 @28198 Added a copy constructor and cleaned up the
** formatting to meet coding standards.
** 005 GES 20060801 @28288 Converted to Externalizable to improve
** efficiency because Serializable has unneeded
** processing (for versioning...) that is
** executed on "inflation".
** 006 GES 20070111 @31786 Moved message types to separate interface
** so calling code can more easily reference
** them. Added documentation.
** 007 GES 20070115 @31833 Added support for conversation mode.
** 008 ECF 20071115 @35882 Refactored net package. isRouterRequest() now
** considers new INIT_ROUTER message type.
** 009 CA 20130822 Fixed the net protocol to support async requests.
** 010 IAS 20200922 Added helper methods for Byteman, changes payload serialization/
** 011 IAS 20200930 Added JMX counters.
*
** 012 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
** 013 HC 20240222 Enabled JMX on FWD Client.
** 014 CA 20240809 Skip using JMX timers when JMX_DEBUG flag is not set.
*/
/*
** This program is free software: you can redistribute it and/or modify
** it under the terms of the GNU Affero General Public License as
** published by the Free Software Foundation, either version 3 of the
** License, or (at your option) any later version.
**
** This program is distributed in the hope that it will be useful,
** but WITHOUT ANY WARRANTY; without even the implied warranty of
** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
** GNU Affero General Public License for more details.
**
** You may find a copy of the GNU Affero GPL version 3 at the following
** location: https://www.gnu.org/licenses/agpl-3.0.en.html
**
** Additional terms under GNU Affero GPL version 3 section 7:
**
** Under Section 7 of the GNU Affero GPL version 3, the following additional
** terms apply to the works covered under the License. These additional terms
** are non-permissive additional terms allowed under Section 7 of the GNU
** Affero GPL version 3 and may not be removed by you.
**
** 0. Attribution Requirement.
**
** You must preserve all legal notices or author attributions in the covered
** work or Appropriate Legal Notices displayed by works containing the covered
** work. You may not remove from the covered work any author or developer
** credit already included within the covered work.
**
** 1. No License To Use Trademarks.
**
** This license does not grant any license or rights to use the trademarks
** Golden Code, FWD, any Golden Code or FWD logo, or any other trademarks
** of Golden Code Development Corporation. You are not authorized to use the
** name Golden Code, FWD, or the names of any author or contributor, for
** publicity purposes without written authorization.
**
** 2. No Misrepresentation of Affiliation.
**
** You may not represent yourself as Golden Code Development Corporation or FWD.
**
** You may not represent yourself for publicity purposes as associated with
** Golden Code Development Corporation, FWD, or any author or contributor to
** the covered work, without written authorization.
**
** 3. No Misrepresentation of Source or Origin.
**
** You may not represent the covered work as solely your work. All modified
** versions of the covered work must be marked in a reasonable way to make it
** clear that the modified work is not originating from Golden Code Development
** Corporation or FWD. All modified versions must contain the notices of
** attribution required in this license.
*/
package com.goldencode.p2j.net;
import java.io.*;
import java.util.logging.*;
import com.goldencode.p2j.jmx.*;
import com.goldencode.p2j.util.logging.*;
import com.goldencode.util.*;
/**
* This class represents messages which are used to exchange information via
* the distributed application protocol.
* <p>
* Each message has a type ({@link MessageTypes}) which defines the how the
* message is to be treated.
* <p>
* Synchronous uses of the protocol (e.g. for a remote object or remote
* procedure call implementation) us the request ID field. This is an
* integer code used to match requests and replies.
* <p>
* There is always a {@link RoutingKey} which specifies the destination.
* <p>
* There is a <code>Serializable</code> (or <code>Externalizable</code>)
* payload for each message. This should not be <code>null</code> but to
* simulate a <code>null</code> payload, one can use the following (a zero
* sized array of objects):
* <p>
* <pre>
* new Object[] {}
* </pre>
* <p>
* Implementing your payload completely as an <code>Externalizable</code> is
* highly recommended as it will improve performance.
*/
public class Message
implements MessageTypes,
Externalizable
{
/** timer for measuring (de)serialization time */
private static final NanoTimer NANO_TIMER = NanoTimer.getInstance(
FwdServerJMX.TimeStat.MessagePayloadS11n
);
/** Logger. */
private static final CentralLogger LOG = CentralLogger.get(Message.class);
/** Message type. */
private int type = 0;
/** Request ID (only used with a synchronous request/reply). */
private int requestID = 0;
/** Routing and requested method information. */
private RoutingKey key = null;
/** Message payload. */
private Serializable payload = null;
/**
* Constructs an empty message. Assumes REQUEST_SYNCH message type. A
* valid <code>RoutingKey</code> instance should be assigned to the
* message before it can be sent.
*/
public Message()
{
type = REQUEST_SYNCH;
}
/**
* Constructs an empty message of specified type. A valid
* <code>RoutingKey</code> instance should be assigned to the message
* before it can be sent.
*
* @param type
* Message type.
*/
public Message(int type)
{
this.type = type;
}
/**
* Constructs a complete message.
*
* @param type
* Message type.
* @param requestID
* Synchronous request ID assigned to message.
* @param key
* An instance of <code>RoutingKey</code> which points to
* required entry point/node.
* @param payload
* Message payload.
*/
public Message(int type,
int requestID,
RoutingKey key,
Serializable payload)
{
this.type = type;
this.requestID = requestID;
this.key = new RoutingKey(key);
this.payload = payload;
}
/**
* Constructs a message of specified type with specified
* <code>RoutingKey</code> and payload.
*
* @param type
* Message type.
* @param key
* An instance of <code>RoutingKey</code> which points to
* required entry point/node.
* @param payload
* Message payload.
*/
public Message(int type, RoutingKey key, Serializable payload)
{
this(type, 0, key, payload);
}
/**
* Constructs a message of specified type and with provided payload. A
* valid <code>RoutingKey</code> instance should be assigned to the
* message before it can be sent.
*
* @param type
* Message type.
* @param payload
* Message payload.
*/
public Message(int type, Serializable payload)
{
this(type, 0, null, payload);
}
/**
* Constructs a synchronous request message with specified
* <code>RoutingKey</code> and payload.
*
* @param key
* An instance of <code>RoutingKey</code> which points to
* required entry point/node.
* @param payload
* Message payload.
*/
public Message(RoutingKey key, Serializable payload)
{
this(REQUEST_SYNCH, 0, key, payload);
}
/**
* Copy constructor.
*
* @param msg
* Message to copy. Must not be <code>null</code>.
*/
public Message(Message msg)
{
this.type = msg.type;
this.requestID = msg.requestID;
this.key = new RoutingKey(msg.key);
this.payload = msg.payload;
}
/**
* Get <code>RoutingKey</code> instance assigned to message.
*
* @return The key stored in the message.
*/
public RoutingKey getKey()
{
return key;
}
/**
* Get the message payload.
*
* @return The message payload.
*/
public Serializable getPayload()
{
return payload;
}
/**
* Get the synchronous request ID.
*
* @return The message request ID.
*/
public int getRequestID()
{
return requestID;
}
/**
* Get the message type.
*
* @return The message type.
*/
public int getType()
{
return type;
}
/**
* Determine if the message is one of the types processed by the
* <code>Dispatcher</code>.
*
* @return <code>true</code> if the message must be processed by the
* <code>Dispatcher</code>.
*/
public boolean isDispatcherRequest()
{
return (type == REQUEST_SYNCH || type == REQUEST_ASYNCH);
}
/**
* Check if message is an ECHO request.
*
* @return <code>true</code> if the message has type <code>ECHO</code>.
*/
public boolean isEchoRequest()
{
return (type == ECHO);
}
/**
* Determine if message is a reply (<code>ADDRESS_REPLY, ECHO_REPLY,
* INIT_REPLY, REPLY, REPLY_EXCEPTION</code>).
*
* @return <code>true</code> if the message is any form of reply.
*/
public boolean isReply()
{
return (type >= BEGIN_REPLY && type <= END_REPLY);
}
/**
* Determine if the message is an async reply (the type is {@link MessageTypes#REPLY_ASYNC},
* {@link MessageTypes#REPLY_EXCEPTION_ASYNC} or the routing key is part of the
* {@link RoutingKey#INTERRUPT_SESSION} group).
*
* @return <code>true</code> if the message is an async reply.
*/
public boolean isAsyncReply()
{
return type == REPLY_ASYNC ||
type == REPLY_EXCEPTION_ASYNC ||
key.getGroupID() == RoutingKey.INTERRUPT_SESSION;
}
/**
* Determine if the message is an async request (the type is {@link MessageTypes#REQUEST_ASYNCH}
* or the routing key is part of the {@link RoutingKey#INTERRUPT_SESSION} group).
*
* @return <code>true</code> if the message is an async reply.
*/
public boolean isAsyncRequest()
{
return type == REQUEST_ASYNCH || key.getGroupID() == RoutingKey.INTERRUPT_SESSION;
}
/**
* Determine if the message is one of types which are processed by the
* <code>Router</code>.
*
* @return <code>true</code> if the message must be processed by the
* <code>Router</code>.
*/
public boolean isRouterRequest()
{
return (type == INIT_STANDARD ||
type == INIT_CONVERSATION ||
type == INIT_ROUTER ||
type == ADDRESS_REQUEST);
}
/**
* Assign <code>RoutingKey</code> to the message.
*
* @param key
* New instance of <code>RoutingKey</code>.
*/
public void setKey(RoutingKey key)
{
this.key = new RoutingKey(key);
}
/**
* Assign new payload to the message.
*
* @param payload
* New message payload.
*/
public void setPayload(Serializable payload)
{
this.payload = payload;
}
/**
* Assigns synchronous request ID to message.
*
* @param requestID
* New synchronous request ID.
*/
public void setRequestID(int requestID)
{
this.requestID = requestID;
}
/**
* Change message type to specified one.
*
* @param type
* New message type.
*/
public void setType(int type)
{
this.type = type;
}
/**
* Replacement for the default object reading method. Note that instead of
* retrieving of the "unknown" <code>RoutingKey</code> instance for the
* <code>key</code> it is explicitly instantiated and then its fields are
* restored by calling the appropriate method.
*
* @param in
* The input source from which fields will be restored.
*
* @throws IOException
* In case of I/O errors.
* @throws ClassNotFoundException
* If payload can't be instantiated.
*/
public void readExternal(ObjectInput in)
throws IOException,
ClassNotFoundException
{
type = in.readInt();
requestID = in.readInt();
key = new RoutingKey();
key.readExternal(in);
if (FwdServerJMX.JMX_DEBUG)
{
NANO_TIMER.timer(() -> setPayload(PayloadSerializer.readPayload(in)));
}
else
{
setPayload(PayloadSerializer.readPayload(in));
}
// onRead();
}
/**
* Replacement for the default object writing method. Note that instead of
* saving <code>key</code> as an ordinary object, it's saved by calling
* appropriate method. This can be done because there is no need to save
* type information in the stream so we can save few bytes of packet size.
*
* @param out
* The output destination to which fields will be saved.
*
* @throws IOException
* In case of I/O errors.
*/
public void writeExternal(ObjectOutput out)
throws IOException
{
out.writeInt(type);
out.writeInt(requestID);
key.writeExternal(out);
// onWrite();
if (FwdServerJMX.JMX_DEBUG)
{
NANO_TIMER.timer(() -> PayloadSerializer.writePayload(out, payload));
}
else
{
PayloadSerializer.writePayload(out, payload);
}
}
/*
The following methods can be used with the following Byteman script to trace messages types:
RULE onRead
CLASS com.goldencode.p2j.net.Message
METHOD readExternal
AT EXIT
BIND msg:Message = $0;
ok:boolean = msg.onRead();
IF FALSE
DO debug("");
ENDRULE
RULE onWrite
CLASS com.goldencode.p2j.net.Message
METHOD writeExternal
AT ENTRY
BIND msg:Message = $0;
ok:boolean = msg.onWrite();
IF FALSE
DO debug("");
ENDRULE
*/
/**
* Helper method for Byteman tracing
* @return always <code>true</code>
*/
public boolean onRead()
{
LOG.log(Level.FINEST, "!!!Deserialized payload; type: %s for %s: %s", type, requestID,
PayloadSerializer.dumpObject(payload));
return true;
}
/**
* Helper method for Byteman tracing
* @return always <code>true</code>
*/
public boolean onWrite()
{
LOG.log(Level.FINEST, "!!!Serializing payload; type: %s for %s: %s", type, requestID,
PayloadSerializer.dumpObject(payload));
return true;
}
}