| /* |
| * 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(); |
| } |
| } |
| |
| } |
