/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.catalina.tribes.group;
import java.io.IOException;
import java.io.Serializable;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import org.apache.catalina.tribes.ByteMessage;
import org.apache.catalina.tribes.ChannelException;
import org.apache.catalina.tribes.ChannelInterceptor;
import org.apache.catalina.tribes.ChannelListener;
import org.apache.catalina.tribes.ChannelMessage;
import org.apache.catalina.tribes.ChannelReceiver;
import org.apache.catalina.tribes.ChannelSender;
import org.apache.catalina.tribes.ErrorHandler;
import org.apache.catalina.tribes.Heartbeat;
import org.apache.catalina.tribes.JmxChannel;
import org.apache.catalina.tribes.ManagedChannel;
import org.apache.catalina.tribes.Member;
import org.apache.catalina.tribes.MembershipListener;
import org.apache.catalina.tribes.MembershipService;
import org.apache.catalina.tribes.RemoteProcessException;
import org.apache.catalina.tribes.UniqueId;
import org.apache.catalina.tribes.group.interceptors.MessageDispatchInterceptor;
import org.apache.catalina.tribes.io.BufferPool;
import org.apache.catalina.tribes.io.ChannelData;
import org.apache.catalina.tribes.io.XByteBuffer;
import org.apache.catalina.tribes.jmx.JmxRegistry;
import org.apache.catalina.tribes.util.Arrays;
import org.apache.catalina.tribes.util.Logs;
import org.apache.catalina.tribes.util.StringManager;
import org.apache.juli.logging.Log;
import org.apache.juli.logging.LogFactory;
/**
* The default implementation of a Channel.<br>
* The GroupChannel manages the replication channel. It coordinates message being sent and received with membership
* announcements. The channel has a chain of interceptors that can modify the message or perform other logic.<br>
* It manages a complete group, both membership and replication.
*/
public class GroupChannel extends ChannelInterceptorBase implements ManagedChannel, JmxChannel, GroupChannelMBean {
private static final Log log = LogFactory.getLog(GroupChannel.class);
protected static final StringManager sm = StringManager.getManager(GroupChannel.class);
/**
* Flag to determine if the channel manages its own heartbeat If set to true, the channel will start a local thread
* for the heart beat.
*/
protected boolean heartbeat = true;
/**
* If <code>heartbeat == true</code> then how often do we want this heartbeat to run. The default value is 5000
* milliseconds.
*/
protected long heartbeatSleeptime = 5 * 1000;
/**
* Internal heartbeat future
*/
protected ScheduledFuture<?> heartbeatFuture = null;
protected ScheduledFuture<?> monitorFuture;
/**
* The <code>ChannelCoordinator</code> coordinates the bottom layer components:<br>
* - MembershipService<br>
* - ChannelSender <br>
* - ChannelReceiver<br>
*/
protected final ChannelCoordinator coordinator = new ChannelCoordinator();
/**
* The first interceptor in the interceptor stack. The interceptors are chained in a linked list, so we only need a
* reference to the first one
*/
protected ChannelInterceptor interceptors = null;
/**
* A list of membership listeners that subscribe to membership announcements
*/
protected final List<MembershipListener> membershipListeners = new CopyOnWriteArrayList<>();
/**
* A list of channel listeners that subscribe to incoming messages
*/
protected final List<ChannelListener> channelListeners = new CopyOnWriteArrayList<>();
/**
* If set to true, the GroupChannel will check to make sure that
*/
protected boolean optionCheck = false;
/**
* the name of this channel.
*/
protected String name = null;
/**
* the jmx domain which this channel is registered.
*/
private String jmxDomain = "ClusterChannel";
/**
* the jmx prefix which will be used with channel ObjectName.
*/
private String jmxPrefix = "";
/**
* If set to true, this channel is registered with jmx.
*/
private boolean jmxEnabled = true;
/**
* Executor service.
*/
protected ScheduledExecutorService utilityExecutor = null;
/**
* the ObjectName of this channel.
*/
private ObjectName oname = null;
/**
* Creates a GroupChannel. This constructor will also add the first interceptor in the GroupChannel.<br>
* The first interceptor is always the channel itself.
*/
public GroupChannel() {
addInterceptor(this);
}
@Override
public void addInterceptor(ChannelInterceptor interceptor) {
if (interceptors == null) {
interceptors = interceptor;
interceptors.setNext(coordinator);
interceptors.setPrevious(null);
coordinator.setPrevious(interceptors);
} else {
ChannelInterceptor last = interceptors;
while (last.getNext() != coordinator) {
last = last.getNext();
}
last.setNext(interceptor);
interceptor.setNext(coordinator);
interceptor.setPrevious(last);
coordinator.setPrevious(interceptor);
}
}
/**
* Sends a heartbeat through the interceptor stack.<br>
* Invoke this method from the application on a periodic basis if you have turned off internal heartbeats
* <code>channel.setHeartbeat(false)</code>
*/
@Override
public void heartbeat() {
super.heartbeat();
for (MembershipListener listener : membershipListeners) {
if (listener instanceof Heartbeat) {
((Heartbeat) listener).heartbeat();
}
}
for (ChannelListener listener : channelListeners) {
if (listener instanceof Heartbeat) {
((Heartbeat) listener).heartbeat();
}
}
}
@Override
public UniqueId send(Member[] destination, Serializable msg, int options) throws ChannelException {
return send(destination, msg, options, null);
}
@Override
public UniqueId send(Member[] destination, Serializable msg, int options, ErrorHandler handler)
throws ChannelException {
if (msg == null) {
throw new ChannelException(sm.getString("groupChannel.nullMessage"));
}
XByteBuffer buffer = null;
try {
if (destination == null || destination.length == 0) {
throw new ChannelException(sm.getString("groupChannel.noDestination"));
}
ChannelData data = new ChannelData(true);// generates a unique Id
data.setAddress(getLocalMember(false));
data.setTimestamp(System.currentTimeMillis());
byte[] b;
if (msg instanceof ByteMessage) {
b = ((ByteMessage) msg).getMessage();
options = options | SEND_OPTIONS_BYTE_MESSAGE;
} else {
b = XByteBuffer.serialize(msg);
options = options & (~SEND_OPTIONS_BYTE_MESSAGE);
}
data.setOptions(options);
// XByteBuffer buffer = new XByteBuffer(b.length+128,false);
buffer = BufferPool.getBufferPool().getBuffer(b.length + 128, false);
buffer.append(b, 0, b.length);
data.setMessage(buffer);
InterceptorPayload payload = null;
if (handler != null) {
payload = new InterceptorPayload();
payload.setErrorHandler(handler);
}
getFirstInterceptor().sendMessage(destination, data, payload);
if (Logs.MESSAGES.isTraceEnabled()) {
Logs.MESSAGES.trace("GroupChannel - Sent msg:" + new UniqueId(data.getUniqueId()) + " at " +
new java.sql.Timestamp(System.currentTimeMillis()) + " to " + Arrays.toNameString(destination));
Logs.MESSAGES.trace("GroupChannel - Send Message:" + new UniqueId(data.getUniqueId()) + " is " + msg);
}
return new UniqueId(data.getUniqueId());
} catch (RuntimeException | IOException e) {
throw new ChannelException(e);
} finally {
if (buffer != null) {
BufferPool.getBufferPool().returnBuffer(buffer);
}
}
}
/**
* Callback from the interceptor stack. <br>
* When a message is received from a remote node, this method will be invoked by the previous interceptor.<br>
* This method can also be used to send a message to other components within the same application, but it's an
* extreme case, and you're probably better off doing that logic between the applications itself.
*
* @param msg ChannelMessage
*/
@Override
public void messageReceived(ChannelMessage msg) {
if (msg == null) {
return;
}
try {
if (Logs.MESSAGES.isTraceEnabled()) {
Logs.MESSAGES.trace("GroupChannel - Received msg:" + new UniqueId(msg.getUniqueId()) + " at " +
new java.sql.Timestamp(System.currentTimeMillis()) + " from " + msg.getAddress().getName());
}
Serializable fwd;
if ((msg.getOptions() & SEND_OPTIONS_BYTE_MESSAGE) == SEND_OPTIONS_BYTE_MESSAGE) {
fwd = new ByteMessage(msg.getMessage().getBytes());
} else {
try {
fwd = XByteBuffer.deserialize(msg.getMessage().getBytesDirect(), 0, msg.getMessage().getLength());
} catch (Exception e) {
log.error(sm.getString("groupChannel.unable.deserialize", msg), e);
return;
}
}
if (Logs.MESSAGES.isTraceEnabled()) {
Logs.MESSAGES.trace("GroupChannel - Receive Message:" + new UniqueId(msg.getUniqueId()) + " is " + fwd);
}
// get the actual member with the correct alive time
Member source = msg.getAddress();
boolean rx = false;
boolean delivered = false;
for (ChannelListener channelListener : channelListeners) {
if (channelListener != null && channelListener.accept(fwd, source)) {
channelListener.messageReceived(fwd, source);
delivered = true;
// if the message was accepted by an RPC channel, that channel
// is responsible for returning the reply, otherwise we send an absence reply
if (channelListener instanceof RpcChannel) {
rx = true;
}
}
} // for
if ((!rx) && (fwd instanceof RpcMessage)) {
// if we have a message that requires a response,
// but none was given, send back an immediate one
sendNoRpcChannelReply((RpcMessage) fwd, source);
}
if (Logs.MESSAGES.isTraceEnabled()) {
Logs.MESSAGES.trace("GroupChannel delivered[" + delivered + "] id:" + new UniqueId(msg.getUniqueId()));
}
} catch (Exception e) {
// this could be the channel listener throwing an exception, we should log it
// as a warning.
if (log.isWarnEnabled()) {
log.warn(sm.getString("groupChannel.receiving.error"), e);
}
throw new RemoteProcessException(sm.getString("groupChannel.receiving.error"), e);
}
}
/**
* Sends a <code>NoRpcChannelReply</code> message to a member<br>
* This method gets invoked by the channel if an RPC message comes in and no channel listener accepts the message.
* This avoids timeout
*
* @param msg RpcMessage
* @param destination Member - the destination for the reply
*/
protected void sendNoRpcChannelReply(RpcMessage msg, Member destination) {
try {
// avoid circular loop
if (msg instanceof RpcMessage.NoRpcChannelReply) {
return;
}
RpcMessage.NoRpcChannelReply reply = new RpcMessage.NoRpcChannelReply(msg.rpcId, msg.uuid);
send(new Member[] { destination }, reply, SEND_OPTIONS_ASYNCHRONOUS);
} catch (Exception e) {
log.error(sm.getString("groupChannel.sendFail.noRpcChannelReply"), e);
}
}
/**
* memberAdded gets invoked by the interceptor below the channel and the channel will broadcast it to the membership
* listeners
*
* @param member Member - the new member
*/
@Override
public void memberAdded(Member member) {
// notify upwards
for (MembershipListener membershipListener : membershipListeners) {
if (membershipListener != null) {
membershipListener.memberAdded(member);
}
}
}
/**
* memberDisappeared gets invoked by the interceptor below the channel and the channel will broadcast it to the
* membership listeners
*
* @param member Member - the member that left or crashed
*/
@Override
public void memberDisappeared(Member member) {
// notify upwards
for (MembershipListener membershipListener : membershipListeners) {
if (membershipListener != null) {
membershipListener.memberDisappeared(member);
}
}
}
/**
* Sets up the default implementation interceptor stack if no interceptors have been added
*
* @throws ChannelException Cluster error
*/
protected synchronized void setupDefaultStack() throws ChannelException {
if (getFirstInterceptor() != null && ((getFirstInterceptor().getNext() instanceof ChannelCoordinator))) {
addInterceptor(new MessageDispatchInterceptor());
}
Iterator<ChannelInterceptor> interceptors = getInterceptors();
while (interceptors.hasNext()) {
ChannelInterceptor channelInterceptor = interceptors.next();
channelInterceptor.setChannel(this);
}
coordinator.setChannel(this);
}
/**
* Validates the option flags that each interceptor is using and reports an error if two interceptor share the same
* flag.
*
* @throws ChannelException Error with option flag
*/
protected void checkOptionFlags() throws ChannelException {
StringBuilder conflicts = new StringBuilder();
ChannelInterceptor first = interceptors;
while (first != null) {
int flag = first.getOptionFlag();
if (flag != 0) {
ChannelInterceptor next = first.getNext();
while (next != null) {
int nflag = next.getOptionFlag();
if (nflag != 0 && (((flag & nflag) == flag) || ((flag & nflag) == nflag))) {
conflicts.append('[');
conflicts.append(first.getClass().getName());
conflicts.append(':');
conflicts.append(flag);
conflicts.append(" == ");
conflicts.append(next.getClass().getName());
conflicts.append(':');
conflicts.append(nflag);
conflicts.append("] ");
} // end if
next = next.getNext();
} // while
} // end if
first = first.getNext();
} // while
if (!conflicts.isEmpty()) {
throw new ChannelException(sm.getString("groupChannel.optionFlag.conflict", conflicts.toString()));
}
}
protected boolean ownExecutor = false;
@Override
public synchronized void start(int svc) throws ChannelException {
setupDefaultStack();
if (optionCheck) {
checkOptionFlags();
}
// register jmx
JmxRegistry jmxRegistry = JmxRegistry.getRegistry(this);
if (jmxRegistry != null) {
this.oname = jmxRegistry.registerJmx(",component=Channel", this);
}
if (utilityExecutor == null) {
log.warn(sm.getString("groupChannel.warn.noUtilityExecutor"));
utilityExecutor = new ScheduledThreadPoolExecutor(1);
ownExecutor = true;
}
super.start(svc);
monitorFuture = utilityExecutor.scheduleWithFixedDelay(this::startHeartbeat, 0, 60, TimeUnit.SECONDS);
}
protected void startHeartbeat() {
if (heartbeat && (heartbeatFuture == null || heartbeatFuture.isDone())) {
if (heartbeatFuture != null && heartbeatFuture.isDone()) {
// There was an error executing the scheduled task, get it and log it
try {
heartbeatFuture.get();
} catch (InterruptedException | ExecutionException e) {
log.error(sm.getString("groupChannel.unable.sendHeartbeat"), e);
}
}
heartbeatFuture = utilityExecutor.scheduleWithFixedDelay(new HeartbeatRunnable(), heartbeatSleeptime,
heartbeatSleeptime, TimeUnit.MILLISECONDS);
}
}
@Override
public synchronized void stop(int svc) throws ChannelException {
if (monitorFuture != null) {
monitorFuture.cancel(true);
monitorFuture = null;
}
if (heartbeatFuture != null) {
heartbeatFuture.cancel(true);
heartbeatFuture = null;
}
super.stop(svc);
if (ownExecutor) {
utilityExecutor.shutdown();
utilityExecutor = null;
ownExecutor = false;
}
if (oname != null) {
JmxRegistry.getRegistry(this).unregisterJmx(oname);
oname = null;
}
}
/**
* Returns the first interceptor of the stack. Useful for traversal.
*
* @return ChannelInterceptor
*/
public ChannelInterceptor getFirstInterceptor() {
return Objects.requireNonNullElse(interceptors, coordinator);
}
@Override
public ScheduledExecutorService getUtilityExecutor() {
return utilityExecutor;
}
@Override
public void setUtilityExecutor(ScheduledExecutorService utilityExecutor) {
this.utilityExecutor = utilityExecutor;
}
@Override
public ChannelReceiver getChannelReceiver() {
return coordinator.getClusterReceiver();
}
@Override
public ChannelSender getChannelSender() {
return coordinator.getClusterSender();
}
@Override
public MembershipService getMembershipService() {
return coordinator.getMembershipService();
}
@Override
public void setChannelReceiver(ChannelReceiver clusterReceiver) {
coordinator.setClusterReceiver(clusterReceiver);
}
@Override
public void setChannelSender(ChannelSender clusterSender) {
coordinator.setClusterSender(clusterSender);
}
@Override
public void setMembershipService(MembershipService membershipService) {
coordinator.setMembershipService(membershipService);
}
@Override
public void addMembershipListener(MembershipListener membershipListener) {
if (!this.membershipListeners.contains(membershipListener)) {
this.membershipListeners.add(membershipListener);
}
}
@Override
public void removeMembershipListener(MembershipListener membershipListener) {
membershipListeners.remove(membershipListener);
}
@Override
public void addChannelListener(ChannelListener channelListener) {
if (!this.channelListeners.contains(channelListener)) {
this.channelListeners.add(channelListener);
} else {
throw new IllegalArgumentException(sm.getString("groupChannel.listener.alreadyExist", channelListener,
channelListener.getClass().getName()));
}
}
@Override
public void removeChannelListener(ChannelListener channelListener) {
channelListeners.remove(channelListener);
}
@Override
public Iterator<ChannelInterceptor> getInterceptors() {
return new InterceptorIterator(this.getNext(), this.coordinator);
}
/**
* Enables/disables the option check<br>
* Setting this to true, will make the GroupChannel perform a conflict check on the interceptors. If two
* interceptors are using the same option flag and throw an error upon start.
*
* @param optionCheck boolean
*/
public void setOptionCheck(boolean optionCheck) {
this.optionCheck = optionCheck;
}
/**
* Configure local heartbeat sleep time<br>
* Only used when <code>getHeartbeat()==true</code>
*
* @param heartbeatSleeptime long - time in milliseconds to sleep between heartbeats
*/
public void setHeartbeatSleeptime(long heartbeatSleeptime) {
this.heartbeatSleeptime = heartbeatSleeptime;
}
/**
* Enables or disables local heartbeat. if <code>setHeartbeat(true)</code> is invoked then the channel will start an
* internal thread to invoke <code>Channel.heartbeat()</code> every <code>getHeartbeatSleeptime</code> milliseconds
*
* @param heartbeat boolean
*/
@Override
public void setHeartbeat(boolean heartbeat) {
this.heartbeat = heartbeat;
}
@Override
public boolean getOptionCheck() {
return optionCheck;
}
@Override
public boolean getHeartbeat() {
return heartbeat;
}
/**
* @return the sleep time in milliseconds that the internal heartbeat will sleep in between invocations of
* <code>Channel.heartbeat()</code>
*/
@Override
public long getHeartbeatSleeptime() {
return heartbeatSleeptime;
}
@Override
public String getName() {
return name;
}
@Override
public void setName(String name) {
this.name = name;
}
@Override
public boolean isJmxEnabled() {
return jmxEnabled;
}
@Override
public void setJmxEnabled(boolean jmxEnabled) {
this.jmxEnabled = jmxEnabled;
}
@Override
public String getJmxDomain() {
return jmxDomain;
}
@Override
public void setJmxDomain(String jmxDomain) {
this.jmxDomain = jmxDomain;
}
@Override
public String getJmxPrefix() {
return jmxPrefix;
}
@Override
public void setJmxPrefix(String jmxPrefix) {
this.jmxPrefix = jmxPrefix;
}
@Override
public ObjectName preRegister(MBeanServer server, ObjectName name) throws Exception {
// NOOP
return null;
}
@Override
public void postRegister(Boolean registrationDone) {
// NOOP
}
@Override
public void preDeregister() throws Exception {
// NOOP
}
@Override
public void postDeregister() {
JmxRegistry.removeRegistry(this, true);
}
/**
* An iterator to loop through the interceptors in a channel.
*/
public static class InterceptorIterator implements Iterator<ChannelInterceptor> {
private final ChannelInterceptor end;
private ChannelInterceptor start;
public InterceptorIterator(ChannelInterceptor start, ChannelInterceptor end) {
this.end = end;
this.start = start;
}
@Override
public boolean hasNext() {
return start != null && start != end;
}
@Override
public ChannelInterceptor next() {
ChannelInterceptor result = null;
if (hasNext()) {
result = start;
start = start.getNext();
}
return result;
}
@Override
public void remove() {
// empty operation
}
}
/**
* <p>
* Title: Internal heartbeat runnable
* </p>
* <p>
* Description: if <code>Channel.getHeartbeat()==true</code> then a thread of this class is created
* </p>
*/
public class HeartbeatRunnable implements Runnable {
@Override
public void run() {
heartbeat();
}
}
}