Queue.java
/*
** Module : Queue.java
** Abstract : message queueing implementation
**
** Copyright (c) 2005-2025, Golden Code Development Corporation.
**
** -#- -I- --Date-- --JPRM-- --------------------------Description-------------------------------
** 001 SIY 20050120 @19502 Created initial version
** 002 SIY 20050207 @19708 Changed stop() and notifyPWT() so now they
** pass an exception which lets waiting threads
** know what did happen. Also updated start()
** and notifyApplication() to accommodate new
** exception delivering scheme.
** 003 SIY 20050211 @19779 Small typo in comments.
** 004 SIY 20050308 @20240 Added termination of the current session in
** stop() method and fixed a lot of small typos.
** 005 NVS 20050308 @20251 Using NetSocket class instead of SSLSocket.
** The ObjectInput and ObjectOutput streams are
** created in NetSocket.
** 006 SIY 20050315 @20386 Organized imports and other minor cleanups.
** 007 SIY 20050325 @20492 Added isContextMatch() method and changed
** access rights for the getNodeAddress() and
** getRemoteAddress() methods.
** 008 SIY 20050329 @20547 Fixed comments, added terminateQueues()
** method.
** 009 NVS 20050418 @20762 Reviewed for repackaging. Fixed comments.
** 010 SIY 20051001 @22920 Added checking of the number of processing
** threads. Fixed comments.
** 011 NVS 20060208 @24475 WaitingThread constructor call changed.
** 012 GES 20060305 @24892 Modified logging code.
** 013 NVS 20060615 @27377 Added a new method kickOff() that initializes
** the connection and the protocol and returns
** without notifying the application.
** 014 GES 20060726 @28200 Added external interface to register state
** synchronizers and made formatting changes to
** match coding standards.
** 015 GES 20060728 @28209 Fixed memory leak in PWT processing,
** simplified waiting thread processing and
** made changes to reduce the 4 calls to
** Control.locate() that occurred for every
** message (2 on each sides of the session).
** 016 GES 20060811 @28567 Moved Control push/pop into transactImpl
** from WaitingThread to eliminate a race
** condition where the enqueued message could
** be transferred to the client and trigger
** a call back before the push even occurred.
** In that case, the Control class' state is
** quite incorrect and the protocol becomes
** unusable.
** 017 NVS 20061017 @30465 Implemented ability to restart the session
** without exiting the process.
** 018 GES 20061019 @30539 Match a minor change in method signature for
** pushRemoteCall().
** 019 GES 20061207 @31613 Made this class safe to call even when the
** session is closed. When an abend occurs
** this class now interrupts any long running
** tasks on the server. Added logging.
** 020 GES 20070111 @31792 Massive rework to simplify and remove the
** over-complicated and often confusing
** notification mechanism. Removed other dead
** code.
** 021 GES 20070115 @31835 Added support for conversation mode.
** 022 GES 20070406 @32850 Fix for CTRL-C in conversation mode. When
** processing was on the peer side, the need
** to use a synchronous calling method for
** an interruption that is asynchronously
** generated, needed special logic to route
** the interruption to the dispatcher instead
** of trying to pass it to the conversation
** thread (which is otherwise occupied).
** 023 GES 20070409 @32868 The sending side of CTRL-C needs to bypass
** the conversation thread in the transactImpl
** and the reply messages for this case
** needed special processing in enqueueInbound
** (both these changes were needed to complete
** the changes started in H022).
** 024 NVS 20070411 @32930 Conversation threads are created as part of a
** thread group named "conversation".
** 025 GES 20071011 @35426 Cleaned up the code for generics, added some
** synchronization and safety code in places.
** 026 ECF 20071101 @35885 Refactored net package. All session-related
** function was moved to the Session hierarchy,
** including context management and event
** notifications. Made class package private.
** Other miscellaneous cleanup.
** 027 GES 20080118 @36861 Added logging code to disable() to allow the
** call path for termination to be logged.
** 028 GES 20081024 @40329 Added methods to hide the socket but still
** provide some essential features.
** 029 ECF 20090205 @42104 Modified stop() method. Added switchContext
** parameter to ensure the active thread has the
** proper security context to interrupt the
** business logic thread and clean up.
** 030 GES 20090722 @43335 Better thread names to enhance debugging.
** 031 GES 20090816 @43635 Massive synchronization updates. At this point,
** all thread safety issues may/should be resolved.
** All core state modifications are now synchronized
** and that synchronization is done in careful
** coordination with other classes like the reader/
** writer threads, the router thread... Note that
** all long running/blocking methods are synchronized
** on a much more granular basis, using volatile
** where possible. It is important for those methods
** to be done this way to avoid deadlocks or
** performance/scaleablity issues. However, that is
** tougher to prove that such approaches are 100%
** correct. With this update, a race condition was
** (between stop() and the session init that occurs
** in the RouterSessionManager) was fixed. In the
** course of fixing that we also cleaned up many
** resource leaks that occurred when that race
** condition triggered.
** 032 CA 20090821 @43717 In transactImpl, when a ApplicationRequestedStop
** error is thrown by the conversation, it will be
** wrapped in a InterruptedException, to allow
** safe propagation.
** 033 CA 20090825 @43754 Fixed a deadlock between Queue and SessionManager.
** This was discovered during remote DB connection
** closing.
** 034 CA 20090831 @43793 Fixed interruption processing when both peers
** are in remote mode. When sending a request, the
** ID of the request is passed to ctrl.pushRemoteCall
** to be pushed on the Control.operations stack.
** 035 CA 20090901 @43813 getControl now receives the context ID, to be
** able to demultiplex the control instance, if
** needed, when this is a server-to-server queue.
** Implemented connection chaining in transactImpl -
** when a request goes out via a virtual session,
** the virtual control is saved as the right-side
** control for the base control (the control for the
** current context).
** 036 CA 20090917 @43928 Fix abnormal connection end. In transactImpl(),
** when a ApplicationRequestedStop is encountered,
** it will be converted to a SilentUnwindExceptions
** so no one can interfere with interruption request.
** 037 CA 20091021 @44181 Any InterruptedException's caught in transactImpl
** will be sent to the interrupt handler.
** 038 CA 20091202 @44471 Rolled back H033 (@43754). The real deadlock was
** with RouterSessionManager.connectVirtual. All
** nested locks should sync on Queue instance first
** and on SessionManager.lock second.
** 039 CA 20130822 Fixed the net protocol to support async requests.
** 040 CA 20131121 The sync'ed state processing is moved from Reader/Writer threads to
** the thread which actually sends/receives the message (i.e.
** Conversation thread).
** 041 CA 20140601 If the queue is not in conversation mode, then always resolve the
** control via Control.locate (do not use the ctrl field).
** 042 IAS 20160331 Removed uneeded synchronization
** 043 GES 20160122 Added safety code to avoid duplicated termination processing.
** 044 GES 20160419 Added termination of the outbound queue at the same time the
** inbound queue is terminated. Make sure that conversation mode
** knows about the thread on which it is executing if a new thread
** is not being started.
** 045 IAS 20160805 Minor changes for the compatibility with new ThreadSafeQueue.
** 046 CA 20170427 When a queue is stopped, context is set explicitly for each session
** associated with this queue, when notifications are sent.
** 047 CA 20180217 Virtual sessions can't cleanup their context concurrently, as they
** might perform remote calls (for e.g. database disconnection) - thus,
** they need to use a common lock.
** 048 GES 20190322 Added isConversationThread() helper.
** 049 IAS 20200722 Refactored to the new NetSocket API.
** VVT 20210917 Javadoc fixed, missing @Override annotations added.
** TJD 20220504 Upgrade do Java 11 minor changes
** 050 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
** 051 GBB 20230825 SecurityManager context & session methods calls updated.
** 052 TJD 20240126 Java 17 fixes, c.g.p2j.net.Queue must be public
** 053 ES 20240618 Prevent throwing a ProtocolViolation exception for an async
** SERVER_INTERRUPT message.
** 054 ES 20250224 Handle the StopConditionException received from the remote-side.
*/
/*
** 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.security.*;
import java.util.*;
import java.net.*;
import java.util.logging.*;
import com.goldencode.p2j.util.*;
import com.goldencode.p2j.util.ErrorManager;
import com.goldencode.p2j.security.SecurityManager;
import com.goldencode.p2j.security.RestrictedUseException;
import com.goldencode.p2j.util.logging.*;
/**
* The Queue class is the kernel of message processing. It coordinates and
* interconnects all classes related to message delivery and processing the
* {@link Dispatcher}, {@link Router} and {@link Protocol}.
*/
public final class Queue
implements LowLevelSession,
MessageTypes
{
/** Logger. */
private static final CentralLogger LOG =
CentralLogger.get(Queue.class.getName());
/** Inbound message queue (shared instance). */
private static final ThreadSafeQueue<InboundRequest> inboundQueue =
new ThreadSafeQueue<InboundRequest>();
/** thread group for the conversation threads */
private static final ThreadGroup convThreads = new ThreadGroup(
Thread.currentThread().getThreadGroup(), "conversation");
/** Outbound message queue (per session). */
private final ThreadSafeQueue<Message> outboundQueue =
new ThreadSafeQueue<Message>();
/** Pool of Waiting Threads. */
private final Map<Integer, WaitingThread> pwt =
new HashMap<Integer, WaitingThread>();
/** Socket which is used for communication with remote side. */
private NetSocket socket = null;
/** Address of the node to which this instance belongs. */
private volatile int nodeAddress = 0;
/** Address of the node at remote side of the connection. */
private volatile int remoteAddress = 0;
/** Instance of the protocol used to send/receive messages. */
private volatile Protocol protocol = null;
/** Last generated request ID. */
private int lastID = 1;
/** Status flag. */
private boolean dead = false;
/** Queue termination exception. */
private Exception exception = null;
/** Interrupt manager. */
private volatile Control ctrl = null;
/** Dedicated conversation mode thread. */
private volatile Conversation conversation = null;
/** <code>true</code> if the session is fully initialized. */
private boolean sessInit = false;
/** A lock used by all sessions which require to cleanup the context. */
private final Object cleanupLock = new Object();
/**
* Constructs an instance.
*
* @param socket
* Low level transport used to communicate with remote side.
*/
Queue(NetSocket socket)
{
this.socket = socket;
}
/**
* Report if the current threads is a conversation thread. A conversation thread is one
* which is dedicated to the processing for a single security session for its lifetime.
* Knowledge of this fact can enable optimizations since the thread is known to never
* change security contexts.
*
* @return {@code true} if the current thread is a conversion thread.
*/
public static boolean isConversationThread()
{
return (Thread.currentThread().getThreadGroup() == convThreads);
}
/**
* Retrieve a message from the Inbound Queue.
*
* @return message instance from the Inbound Queue.
*
* @throws InterruptedException
* is forwarded from <b>ThreadSafeQueue.dequeue() </b>.
*/
static InboundRequest dequeueInbound()
throws InterruptedException
{
return inboundQueue.dequeue();
}
/**
* Terminate inbound queue.
*/
static void terminateQueue()
{
inboundQueue.terminate();
}
/**
* Report if the inbound queue is terminated.
*
* @return <code>true</code> if the inbound queue is terminated.
*/
static boolean isInboundTerminated()
{
return inboundQueue.isTerminated();
}
/**
* Returns the operational state of this session.
*
* @return <code>true</code> if the session is running (operational)
* and <code>false</code> if the session is closed.
*/
@Override
public synchronized boolean isRunning()
{
return !dead && protocol != null && protocol.isRunning();
}
/**
* Sets the instance of the state synchronizer the reader and writer
* threads should use, if and only if there is no other state synchronizer
* already in use. This "there can BE only ONE" approach can be changed
* easily in the future but was implemented to ensure high performance
* since each loop of the protocol must call the synchronization
* primitives.
*
* @param ssync
* The synchronizer to register or <code>null</code> if the
* currently registered synchronizer should be deregistered.
*/
public synchronized void registerSynchronizer(StateSynchronizer ssync)
{
if (protocol != null)
{
protocol.registerSynchronizer(ssync);
}
}
/**
* Return the exception which was passed to the {@link #stop} method, if
* any.
* <p>
* If not <code>null</code>, it will contain the queue termination reason.
*
* @return The queue termination reason. This will be <code>null</code>
* if the queue is not yet terminated or if the queue's
* termination was not abnormal.
*/
@Override
public synchronized Exception getException()
{
return exception;
}
/**
* Provide an access to the node address assigned to the queue.
*
* @return Node address.
*/
@Override
public int getNodeAddress()
{
return nodeAddress;
}
/**
* Provide an access to the remote node address known to the queue.
*
* @return Remote node address.
*/
@Override
public int getRemoteAddress()
{
return remoteAddress;
}
/**
* Get the context cleanup {@link #cleanupLock lock}, to avoid concurrent cleanups.
*
* @return The instance on which to lock for context cleanup.
*/
Object getCleanupLock()
{
return cleanupLock;
}
/**
* Explicitly stop message queue processing, notify all interested parties
* that queue is about to shutdown, and either release all waiting threads
* or halt the current conversation.
*
* @param reason
* The reason for the shutdown or <code>null</code> to signal
* a clean shutdown. Any <code>non-null</code> value indicates
* an abnormal end (abend).
* @param switchContext
* Optionally switch the security context to that of the session's
* user. This is necessary when this method is invoked from
* secondary protocol threads which do not have the security
* context of the user.
*/
synchronized void stop(Exception reason, boolean switchContext)
{
// disable the queue
disable(reason);
SessionManager sessMgr = SessionManager.get();
boolean rtr = sessMgr.isRouter();
boolean hasContext = false;
Object orphanedCtx = null;
try
{
// set security context for user session (when called from secondary
// protocol threads, we don't have the security context of the user)
if (switchContext)
{
hasContext = SecurityManager.getInstance().contextSm.hasContext();
if (!hasContext)
{
sessMgr.setInitialContext();
}
// even if the security mgr has context, there is a short window
// of time between security mgr authentication (when it's session
// is created) and the call by the router to RouterSessionManager's
// createDirectSession() which finishes necessary setup of the
// RSM state such that this setContext() call can work; if the
// queue is stopping in that window of time we must avoid this next
// call
if (sessInit || !rtr)
{
// do not set the session context here - this is required only when sending
// session termination events, which need to execute in the session's context;
// for virtual queues, is possible to have multiple virtual sessions registered
// on it, so context switch must be executed for each session, individually.
// see SessionManager.notifyTermination
}
else
{
RouterSessionManager rsm = (RouterSessionManager) sessMgr;
orphanedCtx = rsm.abortSessionInit(this);
}
}
// notify all registered notifiables that the queue is stopping, but
// only if the session was already initialized
if (sessInit)
sessMgr.notifyTermination(this);
// stop protocol and clean up transport
stopTransport(reason);
// notify the session manager that this queue has stopped, so that
// any associated sessions can be cleaned up, but only if the session
// was already initialized
if (sessInit)
sessMgr.queueStopped(this);
}
finally
{
if (switchContext)
{
if (!hasContext)
{
// back out temporary initial context
sessMgr.dropInitialContext();
}
}
if (rtr && !sessInit && orphanedCtx != null)
{
try
{
// cleanup final security manager context since it will not be
// called by the session.end() code (since there never was a
// session)
SecurityManager.getInstance().sessionSm.terminateSession(orphanedCtx);
}
catch (RestrictedUseException rue)
{
// there is nothing more we can do here, ignore it
}
}
// remember that the session was terminated
sessInit = false;
}
}
/**
* Disable the queue in preparation for shutting it down. Once this
* method is executed, {@link #isRunning()} will report false, even if the
* underlying transport is still active.
* <p>
* This call should be followed by a call to {@link
* #stopTransport(Exception)} to actually shut down the transport
* mechanism.
*
* @param reason
* The reason for the shutdown or <code>null</code> to signal
* a clean shutdown. Any <code>non-null</code> value indicates
* an abnormal end (abend).
*/
synchronized void disable(Exception reason)
{
if (dead)
{
return;
}
if (reason != null)
{
exception = reason;
}
// put message into log
if (LOG.isLoggable(Level.FINER))
{
String lbl = (reason == null) ? "" : " Reason = ";
String msg = (reason == null) ? "" : reason.getMessage();
String err = String.format("Queue is stopping.%s%s", lbl, msg);
// create a stack trace to show where the request came from BUT
// DON'T throw the exception!
String txt = "Only a stack trace, not a real exception!";
Exception exc = new RuntimeException(txt);
LOG.logp(Level.FINER,
"Queue.disable()",
"",
err,
exc);
}
dead = true;
if (SessionManager.get().isRouter())
{
Router.queueStopping(this);
}
}
/**
* Stop the protocol threads and either release all waiting threads or halt
* the current conversation.
* <p>
* This call should be preceded by a call to {@link #disable(Exception)}.
*
* @param reason
* The reason for the shutdown or <code>null</code> to signal
* a clean shutdown. Any <code>non-null</code> value indicates
* an abnormal end (abend).
*/
synchronized void stopTransport(Exception reason)
{
assert !isRunning();
// stop the protocol
if (protocol != null)
{
protocol.stop();
// this will end processing related to the writer thread (which should have
// already been terminated by the call to protocol.stop() above)
outboundQueue.terminate();
}
if (conversation == null)
{
// clear any blocked threads and "return" the exception as the cause
notifyPWT(reason);
}
else
{
// halt the active conversation
conversation.halt();
}
}
/**
* Send provided message as an asynchronous request without waiting for a
* reply. The message type will be forced to <code>REQUEST_ASYNCH</code>.
*
* @param message
* Request message.
*/
public void forward(Message message)
{
checkState();
message.setType(REQUEST_ASYNCH);
enqueueOutbound(message);
}
/**
* Send provided message as a synchronous request and wait for reply.
* Return message payload as a result.
* <p>
* The message passed as a parameter must be completely filled with all
* required information - routing key should point to valid entry point. If
* addresses in the routing key are not set then destination is assumed at
* directly connected node and appropriate addresses are assigned
* automatically.
* <p>
* Caller can specify timeout for waiting reply message. If timeout is set
* to 0 then queue will wait for reply as long as queue is running.
*
* @param message
* Request message.
* @param timeout
* Reply wait timeout.
*
* @return Reply message payload.
*
* @throws Exception
* Is thrown if reply received from remote side has an exception
* as its payload, if a timeout occurs before the reply message
* is received or if a problem occurs during transmission.
*/
public Object transact(Message message, int timeout)
throws Exception
{
checkState();
// if this call is from a registered asynchronous thread, make the request ASYNC
int type = SessionManager.get().isAsyncThread(Thread.currentThread())
? REQUEST_ASYNCH
: REQUEST_SYNCH;
message.setType(type);
Message reply = transactImpl(message, timeout);
if (reply.getType() == REPLY_EXCEPTION_SYNC || reply.getType() == REPLY_EXCEPTION_ASYNC)
{
if (reply.getPayload() instanceof StopConditionException)
{
ErrorManager.handleStopException((StopConditionException) reply.getPayload());
}
throw (Exception) reply.getPayload();
}
return reply.getPayload();
}
/**
* Gets the registered instance of the interrupt manager.
*
* @param contextID
* the security context ID
* @return The interrupt manager or <code>null</code> if no instance
* is registered.
*/
synchronized Control getControl(int contextID)
{
if (contextID != 0 || conversation == null)
return Control.locate(contextID);
if (ctrl == null)
ctrl = Control.locate(contextID);
return ctrl;
}
/**
* Retrieve a message from the Outbound Queue.
*
* @return message instance from the Outbound Queue.
*
* @throws InterruptedException
* is forwarded from <b>ThreadSafeQueue.dequeue() </b>.
*/
Message dequeueOutbound()
throws InterruptedException
{
return outboundQueue.dequeue();
}
/**
* Preliminary message processing method. All incoming messages are
* delivered to the queue (on the protocol's reader thread) via this
* method. Some preliminary processing occurs on this calling thread
* and usually the message is then queued for processing by the router,
* the dispatcher or a waiting thread. In the case of echo processing,
* all processing occurs on the calling thread. Either way, this method
* will not block, but rather it will return quickly with no long running
* synchronous processing.
* <p>
* The algorithm:
* <p>
* <pre>
* Message Type Disposition
* ------------------------------------------- -------------------------
* ADDRESS_REQUEST router
* INIT_STANDARD router
* INIT_CONVERSATION router
* any where destination addr != current addr router
* ECHO immediate send back
* REQUEST_SYNCH dispatcher
* REQUEST_ASYNCH dispatcher
* all forms of REPLY wake waiting thread
* </pre>
*
* @param obj
* The message to process.
*
* @throws ProtocolViolation
* In case of protocol error.
* @throws InterruptedException
* If routing queue or inbound queue is terminated.
*/
void enqueueInbound(Serializable obj)
throws ProtocolViolation,
InterruptedException
{
checkState();
Message msg = (Message) obj;
// ROUTER: must be the first type check so that any type of message can
// be forwarded
if (msg.getKey().getDestinationAddress() != nodeAddress ||
msg.isRouterRequest())
{
Router.forward(new InboundRequest(msg, this));
return;
}
// ECHO support
if (msg.isEchoRequest())
{
SessionManager.get().processEcho(this, msg);
return;
}
int grpid = msg.getKey().getGroupID();
// REPLY (unblock a waiting thread)
if (msg.isReply())
{
// in conversation mode, reply messages (including reply exceptions
// but excluding session interrupt reply) must be routed to the
// conversation thread; note that session interrupt is synchronous
// so that the peer can report on its success in interrupting the
// thread which is why it has a reply message we must handle here;
// since an interrupt occurs at an arbitrary point in the peer's
// logic, the normal conversation mode must be bypassed
if (conversation == null || msg.isAsyncReply())
{
Integer key = Integer.valueOf(msg.getRequestID());
WaitingThread wt = null;
synchronized (pwt)
{
wt = pwt.remove(key);
}
int groupID = msg.getKey().getGroupID();
if (wt != null)
{
wt.notifyMessage(msg);
}
else if (groupID == RoutingKey.SERVER_INTERRUPT)
{
return;
}
else
{
throw new ProtocolViolation("No waiting thread found for " +
"synchronous reply message with " +
"request ID " + key + ".");
}
}
else
{
conversation.reply(msg);
}
return;
}
// REQUEST (synchronous or asynchronous)
if (msg.isDispatcherRequest())
{
// in conversation mode, all synchronous messages (except session
// interrupt) must be routed to the conversation thread; note that
// session interrupt is synchronous so that the peer can report on
// its success in interrupting the thread; since an interrupt occurs
// at an arbitrary point in the peer's logic, the normal conversation
// mode must be bypassed
if (conversation != null &&
msg.getType() == REQUEST_SYNCH &&
grpid != RoutingKey.INTERRUPT_SESSION &&
grpid != RoutingKey.SERVER_INTERRUPT)
{
conversation.request(msg);
}
else
{
// send this to the dispatcher (all asynch messages, all session
// interruptions and any other synchronous messages when not in
// conversation mode)
inboundQueue.enqueue(new InboundRequest(msg, this));
}
return;
}
// protocol error
throw new ProtocolViolation("Unknown message type!");
}
/**
* Put message into Outbound Queue for delivery to the remote side.
*
* @param msg
* An instance of <code>Message</code>.
*
* @throws InvalidParameterException
* if message does not contain routing information.
*/
void enqueueOutbound(Message msg)
throws InvalidParameterException
{
checkState();
RoutingKey key = msg.getKey();
if (key == null)
throw new InvalidParameterException("No routing key in the message");
int dest = key.getDestinationAddress();
if (msg.isReply())
{
if (dest == nodeAddress)
{
key.swapAddresses();
}
if (isLocalAddress(dest))
{
key.setContextID(0);
}
}
else
{
// check validity of the routing key addresses
if (dest == 0)
key.setDestinationAddress(remoteAddress);
if (key.getSourceAddress() == 0)
key.setSourceAddress(nodeAddress);
}
msg = protocol.attachChanges(msg);
try
{
outboundQueue.enqueue(msg);
}
catch (Exception e)
{
throw new InvalidParameterException("Outbound queue is terminated.");
}
}
/**
* Return an instance of <code>AddressPair</code> built using local node
* and remote node addresses. Note that instance is built for use by remote
* side (i.e. remote address and node address are swapped). Refer to
* <code>AddressPair</code> description for more details.
*
* @return An instance of <code>AddressPair</code> built from local node
* and remote node addresses.
*
* @see com.goldencode.p2j.net.AddressPair
*/
AddressPair getAddressPair()
{
return new AddressPair(remoteAddress, nodeAddress);
}
/**
* Report the socket address for the given queue.
*
* @return Socket address of queue.
*/
synchronized InetSocketAddress getRemoteInetAddress()
{
return socket.getRemoteSockAddr();
}
/**
* Reports if the low level networking transport is closed. When closed,
* the transport cannot be used.
*
* @return <code>true</code> if the transport is closed.
*/
synchronized boolean isTransportClosed()
{
return socket.isClosed();
}
/**
* Closes the low-level networking transport.
*/
synchronized void closeTransport()
{
socket.close();
}
/**
* Write data to the socket
*
* @param data
* data to be written
* @throws IOException
* on error
*/
void write(byte[] data)
throws IOException
{
socket.write(data);
}
/**
* Read next portion of input from the socket
*
* @return next portion of input
*
* @throws IOException
* on error
*/
byte[] read()
throws IOException
{
return socket.read();
}
/**
* Get the {@link Protocol} instance of this queue.
*
* @return See above.
*/
synchronized Protocol getProtocol()
{
return protocol;
}
/**
* Assign new node address to the queue.
*
* @param address
* New node address.
*/
void setNodeAddress(int address)
{
nodeAddress = address;
}
/**
* Assign new remote node address to the queue.
*
* @param address
* New remote node address.
*/
void setRemoteAddress(int address)
{
remoteAddress = address;
}
/**
* Start queue. This includes creation of <code>Protocol</code> instance
* and address negotiation if this is requested by application.
* <p>
* All address information related to this queue instance known by the
* application should be assigned before calling this method.
* <p>
* If conversation mode is requested, then all synchronous messages will
* be processed by a single dedicated thread on both sides of the session.
*
* @param init
* <code>true</code> if INIT_STANDARD or INIT_CONVERSATION
* message should be sent to the other side (this invokes node
* address negotiation) for a leaf node, or if INIT_ROUTER
* should be sent for a routing node. If <code>false</code> is
* passed as a parameter then message is not sent and timeout
* parameter is ignored.
* @param timeout
* Timeout waiting for the INIT_REPLY
* @param convo
* <code>true</code> to enable conversation mode.
* @param thread
* <code>true</code> to start a dedicated thread in conversation
* mode. Ignored otherwise.
*
* @throws InterruptedException
* forwarded from the transactImpl() method.
* @throws RequestTimeoutException
* forwarded from the transactImpl() method.
* @throws ProtocolViolation
* in case of invalid reply on INIT message.
* @throws Exception
* forwarded from the transactImpl() method.
*/
void start(boolean init, int timeout, boolean convo, boolean thread)
throws InterruptedException,
RequestTimeoutException,
IllegalStateException,
Exception
{
if (isRunning())
{
String errmsg = "Queue cannot be initialized > 1 time.";
throw new IllegalStateException(errmsg);
}
synchronized (this)
{
if (convo)
{
enableConversation(thread, false);
}
if (protocol == null)
{
protocol = new Protocol(this);
}
protocol.start();
}
if (init)
{
AddressPair pair = new AddressPair();
// honor the node address if it already is set
pair.setNodeAddress(nodeAddress);
int itype;
if (SessionManager.get().isRouter())
{
itype = INIT_ROUTER;
}
else
{
itype = convo ? INIT_CONVERSATION : INIT_STANDARD;
}
Message message = new Message(itype, new RoutingKey(), pair);
Message reply = transactImpl(message, timeout);
Object payload = reply.getPayload();
int type = reply.getType();
if (type == INIT_REPLY && payload instanceof AddressPair)
{
pair = (AddressPair) payload;
synchronized (this)
{
nodeAddress = pair.getNodeAddress();
remoteAddress = pair.getRemoteAddress();
}
}
else
{
throw new ProtocolViolation("Invalid reply on INIT request");
}
}
}
/**
* Force the reader and writer threads to have more descriptive names
* to aid debugging. This is only meaningful on a router node.
*/
synchronized void setProtocolNames()
{
checkState();
if (protocol != null)
protocol.setNames();
}
/**
* Underlying method for all other transactXX() methods. It performs all
* necessary steps for the synchronous request processing: generates unique
* request ID, inserts a <code>WaitingThread</code> instance into PWT and
* wait for a reply. Then it returns a reply message or throws an exception
* if reply is not arrived in time.
*
* @param message
* Request message.
* @param timeout
* Reply wait timeout.
*
* @return Reply <code>Message</code> instance.
*
* @throws InterruptedException
* if waiting for the message was interrupted.
* @throws RequestTimeoutException
* if message is not arrived in time.
* @throws Exception
* if it is delivered from the
* <code>WaitingThread.waitMessage()</code>
*/
Message transactImpl(Message message, int timeout)
throws InterruptedException,
RequestTimeoutException,
Exception
{
checkState();
WaitingThread waitingThread = null;
int req = 0;
boolean async = message.isAsyncRequest();
if (conversation == null || async)
{
synchronized (pwt)
{
Integer key = null;
waitingThread = new WaitingThread();
// find the next available request ID (one that is not in the map)
do
{
// this is safe because even when lastID wraps it will still
// be a valid ID
req = lastID++;
key = Integer.valueOf(req);
}
while (pwt.containsKey(key));
pwt.put(key, waitingThread);
}
}
else
{
// this is a safety measure only, in conversation mode there really
// is no way to get out of synch nor do we really need a request ID
// since they will always be processed in a LIFO order
req = lastID++;
}
message.setRequestID(req);
Control ctrl = getControl(message.getKey().getContextID());
Message reply = null;
if (!async)
{
// this must bracket the message enqueue to ensure that any
// recursive call back OR interruption encounters the proper state
ctrl.pushRemoteCall(req);
}
// this handles the actual sending of the message (and processing on
// the other side)
enqueueOutbound(message);
Control base = Control.locate(0);
Control saved = base.getRightSideControl();
try
{
base.setRightSideControl(ctrl);
if (conversation == null || async)
{
// wait for the response, if an exception occurs on the other
// side, it will be honored (by re-throwing it) within this call
reply = waitingThread.waitMessage(timeout);
}
else
{
// this will never be null as conversation mode doesn't support
// timeouts
reply = conversation.waitMessage(req);
}
}
catch (InterruptedException ie)
{
// the InterruptedException can be thrown while the queue is waiting
// for a notification on the waitingThread - this must sent to the
// custom interruption handler
Control.handleInterrupted(ie);
}
catch (ApplicationRequestedStop ars)
{
throw new SilentUnwindException(ars);
}
finally
{
base.setRightSideControl(saved);
if (!async)
{
// bottom part of the bracket, cleanup state and honor
// interruption if needed
ctrl.popRemoteCall();
}
}
if (reply == null)
throw new RequestTimeoutException("Transaction timeout!");
// post-processing of the message
reply = protocol.applyChanges(reply);
return reply;
}
/**
* Enable conversation mode. This will shift the queue into conversation
* mode and may be called during startup or just after startup. If checking
* is enabled and the queue is not running, this method will return as a
* NOP.
*
* @param start
* <code>true</code> if a new thread should be started for the
* conversation.
* @param check
* <code>true</code> if the queue should be checked to confirm
* that it is running. <code>false</code> to bypass that check.
*/
synchronized void enableConversation(boolean start, boolean check)
{
if (check && !isRunning())
return;
// in the start == false case, the current thread is going to be the conversation thread
// AND Conversation.run() will never be called, so we have to provide access to the
// thread at construction; for the start == true case we should pass null and let the
// code in Conversation.run() automatically pick up the thread
conversation = new Conversation(this, start ? null : Thread.currentThread());
// start a dedicated thread if needed
if (start)
{
Thread t = new Thread(convThreads, conversation);
t.setDaemon(true);
t.start();
}
}
/**
* Set the value of the session initialized flag to <code>true</code>.
*/
synchronized void sessionInitialized()
{
if (isRunning())
sessInit = true;
}
/**
* Utility method which checks if address belongs to the same address space
* as the local node address.
*
* @param addr
* Address to check.
*
* @return <code>true</code> if address passed as a parameter and local
* node address belong to the same address space (i.e. owned by
* the same routing node).
*/
private boolean isLocalAddress(int addr)
{
return ((addr & AddressServer.ROUTER_MASK) ==
(nodeAddress & AddressServer.ROUTER_MASK));
}
/**
* Release all waiting threads in the PWT and pass an exception to let the
* waiting threads know the disposition of the protocol.
*
* @param reason
* Exception object which will be passed to all waiting threads.
*/
private void notifyPWT(Exception reason)
{
WaitingThread[] threads = null;
synchronized (pwt)
{
threads = pwt.values().toArray(new WaitingThread[0]);
pwt.clear();
}
if (threads != null)
{
if (reason == null)
reason = new ApplicationRequestedStop();
for (int i = 0; i < threads.length; i++)
threads[i].notifyException(reason);
}
}
/**
* Check if the queue is running and throw an exception if it is not.
*
* @throws IllegalStateException
* If the queue is not running.
*/
private synchronized void checkState()
throws IllegalStateException
{
if (!isRunning())
{
IllegalStateException ise =
new IllegalStateException("The queue is not running!");
if (exception != null)
ise.initCause(exception);
throw ise;
}
}
}