// Path: core/src/main/java/io/bigio/Interceptor.java
// public interface Interceptor {
// 
//     /**
//      * Intercept a message.
//      * 
//      * @param envelope the message.
//      * @return a transformation of the received message.
//      */
//     public Envelope intercept(Envelope envelope);
// }
// 
// Path: core/src/main/java/io/bigio/MessageListener.java
// public interface MessageListener<T> {
// 
//     /**
//      * Receive a message.
//      * 
//      * @param message a message.
//      */
//     public void receive(T message);
// }
// 
// Path: core/src/main/java/io/bigio/core/member/Member.java
// public interface Member {
//     
//     public AtomicInteger getSequence();
// 
//     public MemberStatus getStatus();
// 
//     public void setStatus(MemberStatus status);
// 
//     public Map<String, String> getTags();
// 
//     public String getIp();
// 
//     public void setIp(String ip);
// 
//     public int getDataPort();
// 
//     public void setDataPort(int dataPort);
// 
//     public int getGossipPort();
// 
//     public void setGossipPort(int gossipPort);
// 
//     public void send(Envelope message) throws IOException;
// 
//     public byte[] getPublicKey();
// 
//     public void setPublicKey(byte[] publicKey);
// 
// }
// 
// Path: core/src/main/java/io/bigio/core/member/MemberKey.java
// public class MemberKey {
// 
//     private MemberKey() {
// 
//     }
// 
//     public static String getKey(Member member) {
//         return new StringBuilder()
//                 .append(member.getIp())
//                 .append(":")
//                 .append(member.getGossipPort())
//                 .append(":")
//                 .append(member.getDataPort()).toString();
//     }
// 
//     public static String getKey(GossipMessage message) {
//         return new StringBuilder()
//                 .append(message.getIp())
//                 .append(":")
//                 .append(message.getGossipPort())
//                 .append(":")
//                 .append(message.getDataPort()).toString();
//     }
// }
// 
// Path: core/src/main/java/io/bigio/util/TopicUtils.java
// public class TopicUtils {
// 
//     private TopicUtils() {
//         
//     }
// 
//     public static final String ALL_PARTITIONS = ".*";
//     
//     public static String getTopicString(String topic, String partition) {
//         return new StringBuilder().append(topic).append("(").append(partition).append(")").toString();
//     }
// 
//     public static String getNotifyTopicString(String topic, String partition) {
//         return new StringBuilder().append(topic).append(partition).toString();
//     }
// 
//     public static String getTopic(String topicPartition) {
//         return topicPartition.split("\\(")[0];
//     }
// 
//     public static String getPartition(String topicPartition) {
//         String[] spl = topicPartition.split("\\(");
//         if(spl.length > 1) {
//             return spl[1];
//         } 
// 
//         return ALL_PARTITIONS;
//     }
// }


import io.bigio.Component;
import io.bigio.Interceptor;
import io.bigio.MessageListener;
import io.bigio.core.member.Member;
import io.bigio.core.member.MemberKey;
import io.bigio.util.TopicUtils;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.Environment;
import reactor.core.Reactor;
import reactor.core.spec.Reactors;
import reactor.event.Event;
import reactor.event.selector.Selectors;
import reactor.function.Consumer;
     * @param topic a topic.
     * @param partition a partition.
     * @param member a member.
     */
    protected synchronized void registerMemberForTopic(String topic, String partition, Member member) {

        if(map.get(member) == null) {
            map.put(member, new ConcurrentHashMap<>());
        }

        if(map.get(member).get(topic) == null) {
            map.get(member).put(topic, new ArrayList<>());
        }

        boolean found = false;
        for(Registration reg : map.get(member).get(topic)) {
            if(topic.equals(reg.getTopic()) && partition.equals(reg.getPartition()) && member.equals(reg.getMember())) {
                found = true;
                break;
            }
        }

        if(!found) {
            Registration reg = new Registration(member, topic, partition);
            map.get(member).get(topic).add(reg);

            if(LOG.isTraceEnabled()) {
                LOG.trace(new StringBuilder()
                        .append(MemberKey.getKey(me))
                        .append(": Registering member '")
// Path: core/src/main/java/io/bigio/Interceptor.java
// public interface Interceptor {
// 
//     /**
//      * Intercept a message.
//      * 
//      * @param envelope the message.
//      * @return a transformation of the received message.
//      */
//     public Envelope intercept(Envelope envelope);
// }
// 
// Path: core/src/main/java/io/bigio/MessageListener.java
// public interface MessageListener<T> {
// 
//     /**
//      * Receive a message.
//      * 
//      * @param message a message.
//      */
//     public void receive(T message);
// }
// 
// Path: core/src/main/java/io/bigio/core/member/Member.java
// public interface Member {
//     
//     public AtomicInteger getSequence();
// 
//     public MemberStatus getStatus();
// 
//     public void setStatus(MemberStatus status);
// 
//     public Map<String, String> getTags();
// 
//     public String getIp();
// 
//     public void setIp(String ip);
// 
//     public int getDataPort();
// 
//     public void setDataPort(int dataPort);
// 
//     public int getGossipPort();
// 
//     public void setGossipPort(int gossipPort);
// 
//     public void send(Envelope message) throws IOException;
// 
//     public byte[] getPublicKey();
// 
//     public void setPublicKey(byte[] publicKey);
// 
// }
// 
// Path: core/src/main/java/io/bigio/core/member/MemberKey.java
// public class MemberKey {
// 
//     private MemberKey() {
// 
//     }
// 
//     public static String getKey(Member member) {
//         return new StringBuilder()
//                 .append(member.getIp())
//                 .append(":")
//                 .append(member.getGossipPort())
//                 .append(":")
//                 .append(member.getDataPort()).toString();
//     }
// 
//     public static String getKey(GossipMessage message) {
//         return new StringBuilder()
//                 .append(message.getIp())
//                 .append(":")
//                 .append(message.getGossipPort())
//                 .append(":")
//                 .append(message.getDataPort()).toString();
//     }
// }
// 
// Path: core/src/main/java/io/bigio/util/TopicUtils.java
// public class TopicUtils {
// 
//     private TopicUtils() {
//         
//     }
// 
//     public static final String ALL_PARTITIONS = ".*";
//     
//     public static String getTopicString(String topic, String partition) {
//         return new StringBuilder().append(topic).append("(").append(partition).append(")").toString();
//     }
// 
//     public static String getNotifyTopicString(String topic, String partition) {
//         return new StringBuilder().append(topic).append(partition).toString();
//     }
// 
//     public static String getTopic(String topicPartition) {
//         return topicPartition.split("\\(")[0];
//     }
// 
//     public static String getPartition(String topicPartition) {
//         String[] spl = topicPartition.split("\\(");
//         if(spl.length > 1) {
//             return spl[1];
//         } 
// 
//         return ALL_PARTITIONS;
//     }
// }


// Path: core/src/main/java/io/bigio/core/ListenerRegistry.java
import io.bigio.Component;
import io.bigio.Interceptor;
import io.bigio.MessageListener;
import io.bigio.core.member.Member;
import io.bigio.core.member.MemberKey;
import io.bigio.util.TopicUtils;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.Environment;
import reactor.core.Reactor;
import reactor.core.spec.Reactors;
import reactor.event.Event;
import reactor.event.selector.Selectors;
import reactor.function.Consumer;

     * @param topic a topic.
     * @param partition a partition.
     * @param member a member.
     */
    protected synchronized void registerMemberForTopic(String topic, String partition, Member member) {

        if(map.get(member) == null) {
            map.put(member, new ConcurrentHashMap<>());
        }

        if(map.get(member).get(topic) == null) {
            map.get(member).put(topic, new ArrayList<>());
        }

        boolean found = false;
        for(Registration reg : map.get(member).get(topic)) {
            if(topic.equals(reg.getTopic()) && partition.equals(reg.getPartition()) && member.equals(reg.getMember())) {
                found = true;
                break;
            }
        }

        if(!found) {
            Registration reg = new Registration(member, topic, partition);
            map.get(member).get(topic).add(reg);

            if(LOG.isTraceEnabled()) {
                LOG.trace(new StringBuilder()
                        .append(MemberKey.getKey(me))
                        .append(": Registering member '")
                        .append(MemberKey.getKey(member))