// 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;
        }
    }

    /**
     * Remove topic/partition registrations. 
     * 
     * @param regs a set of registrations.
     */
    public void removeRegistrations(List<Registration> regs) {
        map.values().stream().filter((allRegs) -> (allRegs != null)).forEach((allRegs) -> {
            allRegs.keySet().stream().forEach((key) -> {
                allRegs.get(key).removeAll(regs);
            });
        });
    }

    /**
     * Get all topic/partition registrations. 
     * 
     * @return the list of all registrations.
     */
    public List<Registration> getAllRegistrations() {
        List<Registration> ret = new ArrayList<>();
        
        map.values().stream().filter((allRegs) -> (allRegs != null)).forEach((allRegs) -> {
            allRegs.keySet().stream().forEach((key) -> {
                ret.addAll(allRegs.get(key));
            });
        });

// 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;

        }
    }

    /**
     * Remove topic/partition registrations. 
     * 
     * @param regs a set of registrations.
     */
    public void removeRegistrations(List<Registration> regs) {
        map.values().stream().filter((allRegs) -> (allRegs != null)).forEach((allRegs) -> {
            allRegs.keySet().stream().forEach((key) -> {
                allRegs.get(key).removeAll(regs);
            });
        });
    }

    /**
     * Get all topic/partition registrations. 
     * 
     * @return the list of all registrations.
     */
    public List<Registration> getAllRegistrations() {
        List<Registration> ret = new ArrayList<>();
        
        map.values().stream().filter((allRegs) -> (allRegs != null)).forEach((allRegs) -> {
            allRegs.keySet().stream().forEach((key) -> {
                ret.addAll(allRegs.get(key));
            });
        });

        return ret;