java/org/apache/catalina/tribes/membership/StaticMembershipProvider.java

/*
 * 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.membership;

import java.io.Serial;
import java.io.Serializable;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import org.apache.catalina.tribes.Channel;
import org.apache.catalina.tribes.ChannelException;
import org.apache.catalina.tribes.ChannelException.FaultyMember;
import org.apache.catalina.tribes.ChannelListener;
import org.apache.catalina.tribes.Heartbeat;
import org.apache.catalina.tribes.Member;
import org.apache.catalina.tribes.group.Response;
import org.apache.catalina.tribes.group.RpcCallback;
import org.apache.catalina.tribes.group.RpcChannel;
import org.apache.catalina.tribes.util.Arrays;
import org.apache.catalina.tribes.util.ExceptionUtils;
import org.apache.catalina.tribes.util.StringManager;
import org.apache.juli.logging.Log;
import org.apache.juli.logging.LogFactory;

public class StaticMembershipProvider extends MembershipProviderBase
        implements RpcCallback, ChannelListener, Heartbeat {

    protected static final StringManager sm = StringManager.getManager(StaticMembershipProvider.class);
    private static final Log log = LogFactory.getLog(StaticMembershipProvider.class);

    protected Channel channel;
    protected RpcChannel rpcChannel;
    private String membershipName = null;
    private byte[] membershipId = null;
    protected ArrayList<StaticMember> staticMembers;
    protected int sendOptions = Channel.SEND_OPTIONS_ASYNCHRONOUS;
    protected long expirationTime = 5000;
    protected int connectTimeout = 500;
    protected long rpcTimeout = 3000;
    protected int startLevel = 0;
    // for ping thread
    protected boolean useThread = false;
    protected long pingInterval = 1000;
    protected volatile boolean running = true;
    protected PingThread thread = null;

    @Override
    public void init(Properties properties) throws Exception {
        String expirationTimeStr = properties.getProperty("expirationTime");
        this.expirationTime = Long.parseLong(expirationTimeStr);
        String connectTimeoutStr = properties.getProperty("connectTimeout");
        this.connectTimeout = Integer.parseInt(connectTimeoutStr);
        String rpcTimeouStr = properties.getProperty("rpcTimeout");
        this.rpcTimeout = Long.parseLong(rpcTimeouStr);
        this.membershipName = properties.getProperty("membershipName");
        this.membershipId = membershipName.getBytes(StandardCharsets.ISO_8859_1);
        membership = new Membership(service.getLocalMember(true));
        this.rpcChannel = new RpcChannel(this.membershipId, channel, this);
        this.channel.addChannelListener(this);
        String useThreadStr = properties.getProperty("useThread");
        this.useThread = Boolean.parseBoolean(useThreadStr);
        String pingIntervalStr = properties.getProperty("pingInterval");
        this.pingInterval = Long.parseLong(pingIntervalStr);
    }

    @Override
    public void start(int level) throws Exception {
        if (Channel.MBR_RX_SEQ == (level & Channel.MBR_RX_SEQ)) {
            // no-op
        }
        if (Channel.MBR_TX_SEQ == (level & Channel.MBR_TX_SEQ)) {
            // no-op
        }
        startLevel = (startLevel | level);
        if (startLevel == (Channel.MBR_RX_SEQ | Channel.MBR_TX_SEQ)) {
            startMembership(getAliveMembers(staticMembers.toArray(new Member[0])));
            running = true;
            if (thread == null && useThread) {
                thread = new PingThread();
                thread.setDaemon(true);
                thread.setName("StaticMembership.PingThread[" + this.channel.getName() + "]");
                thread.start();
            }
        }
    }

    @Override
    public boolean stop(int level) throws Exception {
        if (Channel.MBR_RX_SEQ == (level & Channel.MBR_RX_SEQ)) {
            // no-op
        }
        if (Channel.MBR_TX_SEQ == (level & Channel.MBR_TX_SEQ)) {
            // no-op
        }
        startLevel = (startLevel & (~level));
        if (startLevel == 0) {
            running = false;
            if (thread != null) {
                thread.interrupt();
                thread = null;
            }
            if (this.rpcChannel != null) {
                this.rpcChannel.breakdown();
            }
            if (this.channel != null) {
                try {
                    stopMembership(this.getMembers());
                } catch (Throwable t) {
                    ExceptionUtils.handleThrowable(t);
                    // Otherwise ignore
                }
                this.channel.removeChannelListener(this);
                this.channel = null;
            }
            this.rpcChannel = null;
            this.membership.reset();
        }
        return (startLevel == 0);
    }

    protected void startMembership(Member[] members) throws ChannelException {
        if (members.length == 0) {
            return;
        }
        MemberMessage msg = new MemberMessage(membershipId, MemberMessage.MSG_START, service.getLocalMember(true));
        Response[] resp = rpcChannel.send(members, msg, RpcChannel.ALL_REPLY, sendOptions, rpcTimeout);
        if (resp.length > 0) {
            for (Response response : resp) {
                messageReceived(response.getMessage(), response.getSource());
            }
        } else {
            log.warn(sm.getString("staticMembershipProvider.startMembership.noReplies"));
        }
    }

    protected Member setupMember(Member mbr) {
        // no-op
        return mbr;
    }

    protected void memberAdded(Member member) {
        Member mbr = setupMember(member);
        if (membership.memberAlive(mbr)) {
            Runnable r = () -> {
                Thread currentThread = Thread.currentThread();
                String name = currentThread.getName();
                try {
                    currentThread.setName("StaticMembership-memberAdded");
                    membershipListener.memberAdded(mbr);
                } finally {
                    currentThread.setName(name);
                }
            };
            executor.execute(r);
        }
    }

    protected void memberDisappeared(Member member) {
        membership.removeMember(member);
        Runnable r = () -> {
            Thread currentThread = Thread.currentThread();
            String name = currentThread.getName();
            try {
                currentThread.setName("StaticMembership-memberDisappeared");
                membershipListener.memberDisappeared(member);
            } finally {
                currentThread.setName(name);
            }
        };
        executor.execute(r);
    }

    protected void memberAlive(Member member) {
        if (!membership.contains(member)) {
            memberAdded(member);
        }
        membership.memberAlive(member);
    }

    protected void stopMembership(Member[] members) {
        if (members.length == 0) {
            return;
        }
        Member localmember = service.getLocalMember(false);
        localmember.setCommand(Member.SHUTDOWN_PAYLOAD);
        MemberMessage msg = new MemberMessage(membershipId, MemberMessage.MSG_STOP, localmember);
        try {
            channel.send(members, msg, sendOptions);
        } catch (ChannelException e) {
            log.error(sm.getString("staticMembershipProvider.stopMembership.sendFailed"), e);
        }
    }

    @Override
    public void messageReceived(Serializable msg, Member sender) {
        MemberMessage memMsg = (MemberMessage) msg;
        Member member = memMsg.getMember();
        if (memMsg.getMsgtype() == MemberMessage.MSG_START) {
            memberAdded(member);
        } else if (memMsg.getMsgtype() == MemberMessage.MSG_STOP) {
            memberDisappeared(member);
        } else if (memMsg.getMsgtype() == MemberMessage.MSG_PING) {
            memberAlive(member);
        }
    }

    @Override
    public boolean accept(Serializable msg, Member sender) {
        boolean result = false;
        if (msg instanceof MemberMessage) {
            result = Arrays.equals(this.membershipId, ((MemberMessage) msg).getMembershipId());
        }
        return result;
    }

    @Override
    public Serializable replyRequest(Serializable msg, final Member sender) {
        if (!(msg instanceof MemberMessage memMsg)) {
            return null;
        }
        if (memMsg.getMsgtype() == MemberMessage.MSG_START) {
            messageReceived(memMsg, sender);
            memMsg.setMember(service.getLocalMember(true));
            return memMsg;
        } else if (memMsg.getMsgtype() == MemberMessage.MSG_PING) {
            messageReceived(memMsg, sender);
            memMsg.setMember(service.getLocalMember(true));
            return memMsg;
        } else {
            // other messages are ignored.
            if (log.isInfoEnabled()) {
                log.info(sm.getString("staticMembershipProvider.replyRequest.ignored", memMsg.getTypeDesc()));
            }
            return null;
        }
    }

    @Override
    public void leftOver(Serializable msg, Member sender) {
        if (!(msg instanceof MemberMessage memMsg)) {
            return;
        }
        if (memMsg.getMsgtype() == MemberMessage.MSG_START) {
            messageReceived(memMsg, sender);
        } else if (memMsg.getMsgtype() == MemberMessage.MSG_PING) {
            messageReceived(memMsg, sender);
        } else {
            // other messages are ignored.
            if (log.isInfoEnabled()) {
                log.info(sm.getString("staticMembershipProvider.leftOver.ignored", memMsg.getTypeDesc()));
            }
        }
    }

    @Override
    public void heartbeat() {
        try {
            if (!useThread) {
                ping();
            }
        } catch (ChannelException e) {
            log.warn(sm.getString("staticMembershipProvider.heartbeat.failed"), e);
        }
    }

    protected void ping() throws ChannelException {
        // send ping
        Member[] members = getAliveMembers(staticMembers.toArray(new Member[0]));
        if (members.length > 0) {
            try {
                MemberMessage msg =
                        new MemberMessage(membershipId, MemberMessage.MSG_PING, service.getLocalMember(true));
                Response[] resp = rpcChannel.send(members, msg, RpcChannel.ALL_REPLY, sendOptions, rpcTimeout);
                for (Response response : resp) {
                    messageReceived(response.getMessage(), response.getSource());
                }
            } catch (ChannelException ce) {
                // Handle known failed members
                FaultyMember[] faultyMembers = ce.getFaultyMembers();
                for (FaultyMember faultyMember : faultyMembers) {
                    memberDisappeared(faultyMember.getMember());
                }
                throw ce;
            }
        }
        // expire
        checkExpired();
    }

    protected void checkExpired() {
        Member[] expired = membership.expire(expirationTime);
        for (Member member : expired) {
            membershipListener.memberDisappeared(member);
        }
    }

    public void setChannel(Channel channel) {
        this.channel = channel;
    }

    public void setStaticMembers(ArrayList<StaticMember> staticMembers) {
        this.staticMembers = staticMembers;
    }

    private Member[] getAliveMembers(Member[] members) {
        List<Member> aliveMembers = new ArrayList<>();
        for (Member member : members) {
            try (Socket socket = new Socket()) {
                InetAddress ia = InetAddress.getByAddress(member.getHost());
                InetSocketAddress addr = new InetSocketAddress(ia, member.getPort());
                socket.connect(addr, connectTimeout);
                aliveMembers.add(member);
            } catch (Throwable t) {
                ExceptionUtils.handleThrowable(t);
                // Otherwise ignore
            }
        }
        return aliveMembers.toArray(new Member[0]);
    }

    // ------------------------------------------------------------------------------
    // member message to send to and from other memberships
    // ------------------------------------------------------------------------------
    public static class MemberMessage implements Serializable {
        @Serial
        private static final long serialVersionUID = 1L;
        public static final int MSG_START = 1;
        public static final int MSG_STOP = 2;
        public static final int MSG_PING = 3;
        private final int msgtype;
        private final byte[] membershipId;
        private Member member;

        public MemberMessage(byte[] membershipId, int msgtype, Member member) {
            this.membershipId = membershipId;
            this.msgtype = msgtype;
            this.member = member;
        }

        public int getMsgtype() {
            return msgtype;
        }

        public byte[] getMembershipId() {
            return membershipId;
        }

        public Member getMember() {
            return member;
        }

        public void setMember(Member local) {
            this.member = local;
        }

        @Override
        public String toString() {
            return "MemberMessage[" + "name=" + new String(membershipId) + "; type=" + getTypeDesc() + "; member=" +
                    member + ']';
        }

        protected String getTypeDesc() {
            return switch (msgtype) {
                case MSG_START -> "MSG_START";
                case MSG_STOP -> "MSG_STOP";
                case MSG_PING -> "MSG_PING";
                default -> "UNKNOWN";
            };
        }
    }

    protected class PingThread extends Thread {
        @Override
        public void run() {
            while (running) {
                try {
                    sleep(pingInterval);
                    ping();
                } catch (InterruptedException ix) {
                    // Ignore
                } catch (Exception e) {
                    log.warn(sm.getString("staticMembershipProvider.pingThread.failed"), e);
                }
            }
        }
    }
}