public class Message extends BaseMessage implements Serializable {

    private String msgId;
    private String topic;
    private byte[] body;
    private long timeStamp;

    public String getMsgId() {
        return msgId;
    }

    public void setMsgId(String msgId) {
        this.msgId = msgId;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public byte[] getBody() {
        return body;
    }

    public void setBody(byte[] body) {
        this.body = body;
    }

    public long getTimeStamp() {
        return timeStamp;
    }

    public void setTimeStamp(long timeStamp) {
        this.timeStamp = timeStamp;
    }

    public String toString() {
        ReflectionToStringBuilder.setDefaultStyle(ToStringStyle.SHORT_PREFIX_STYLE);
        return ReflectionToStringBuilder.toStringExclude(this, new String[]{"body"});
    }

    public int hashCode() {
        return new HashCodeBuilder(11, 23).append(msgId).toHashCode();
    }

    public boolean equals(Object obj) {
        boolean result = false;
        if (obj != null && Message.class.isAssignableFrom(obj.getClass())) {
            Message msg = (Message) obj;
            result = new EqualsBuilder().append(topic, msg.getTopic()).append(body, msg.getBody())
                    .isEquals();
        }
        return result;
    }
} public class ProducerAckMessage extends BaseMessage implements Serializable {

    private String ack;
    private int status;
    private String msgId;

    public String getAck() {
        return ack;
    }

    public void setAck(String ack) {
        this.ack = ack;
    }

    public int getStatus() {
        return status;
    }

    public void setStatus(int status) {
        this.status = status;
    }

    public String getMsgId() {
        return msgId;
    }

    public void setMsgId(String msgId) {
        this.msgId = msgId;
    }
} public class SemaphoreCache {

    private final static int hookTime = MessageSystemConfig.SemaphoreCacheHookTimeValue;

    private static final LoadingCache<String, Semaphore> cache = CacheBuilder.newBuilder().
            concurrencyLevel(NettyClustersConfig.getWorkerThreads()).
            build(new CacheLoader<String, Semaphore>() {
                public Semaphore load(String input) throws Exception {
                    return new Semaphore(0);
                }
            });

    public static int getAvailablePermits(String key) {
        try {
            return cache.get(key).availablePermits();
        } catch (ExecutionException ex) {
            Logger.getLogger(SemaphoreCache.class.getName()).log(Level.SEVERE, null, ex);
            return 0;
        }
    }

    public static void release(String key) {
        try {
            cache.get(key).release();
            TimeUnit.MILLISECONDS.sleep(hookTime);
        } catch (ExecutionException ex) {
            Logger.getLogger(SemaphoreCache.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InterruptedException ex) {
            Logger.getLogger(SemaphoreCache.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    public static void acquire(String key) {
        try {
            cache.get(key).acquire();
            TimeUnit.MILLISECONDS.sleep(hookTime);
        } catch (InterruptedException ex) {
            Logger.getLogger(SemaphoreCache.class.getName()).log(Level.SEVERE, null, ex);
        } catch (ExecutionException ex) {
            Logger.getLogger(SemaphoreCache.class.getName()).log(Level.SEVERE, null, ex);
        }
    }
} public class ConsumerClusters {

    private int next = 0;
    private final String clustersId;
    private final ConcurrentHashMap<String, SubscriptionData> subMap
            = new ConcurrentHashMap<String, SubscriptionData>();

    private final ConcurrentHashMap<String, RemoteChannelData> channelMap
            = new ConcurrentHashMap<String, RemoteChannelData>();

    private final List<RemoteChannelData> channelList = Collections.synchronizedList(new ArrayList<RemoteChannelData>());

    public ConsumerClusters(String clustersId) {
        this.clustersId = clustersId;
    }

    public String getClustersId() {
        return clustersId;
    }

    public ConcurrentHashMap<String, SubscriptionData> getSubMap() {
        return subMap;
    }

    public ConcurrentHashMap<String, RemoteChannelData> getChannelMap() {
        return channelMap;
    }

    public void attachRemoteChannelData(String clientId, RemoteChannelData channelinfo) {
        if (findRemoteChannelData(channelinfo.getClientId()) == null) {
            channelMap.put(clientId, channelinfo);
            subMap.put(channelinfo.getSubcript().getTopic(), channelinfo.getSubcript());
            channelList.add(channelinfo);
        } else {
            System.out.println("consumer clusters exists! it's clientId:" + clientId);
        }
    }

    public void detachRemoteChannelData(String clientId) {
        channelMap.remove(clientId);

        Predicate predicate = new Predicate() {
            public boolean evaluate(Object object) {
                String id = ((RemoteChannelData) object).getClientId();
                return id.compareTo(clientId) == 0;
            }
        };

        RemoteChannelData data = (RemoteChannelData) CollectionUtils.find(channelList, predicate);
        if (data != null) {
            channelList.remove(data);
        }
    }

    public RemoteChannelData findRemoteChannelData(String clientId) {
        return (RemoteChannelData) MapUtils.getObject(channelMap, clientId);
    }

    public RemoteChannelData nextRemoteChannelData() {

        Predicate predicate = new Predicate() {
            public boolean evaluate(Object object) {
                RemoteChannelData data = (RemoteChannelData) object;
                Channel channel = data.getChannel();
                return NettyUtil.validateChannel(channel);
            }
        };

        CollectionUtils.filter(channelList, predicate);
        return channelList.get(next++ % channelList.size());
    }

    public SubscriptionData findSubscriptionData(String topic) {
        return this.subMap.get(topic);
    }
} public class ConsumerContext {

    private static final CopyOnWriteArrayList<ClustersRelation> relationArray = new CopyOnWriteArrayList<ClustersRelation>();
    private static final CopyOnWriteArrayList<ClustersState> stateArray = new CopyOnWriteArrayList<ClustersState>();

    public static void setClustersStat(String clusters, int stat) {
        stateArray.add(new ClustersState(clusters, stat));
    }

    public static int getClustersStat(String clusters) {

        Predicate predicate = new Predicate() {
            public boolean evaluate(Object object) {
                String clustersId = ((ClustersState) object).getClusters();
                return clustersId.compareTo(clusters) == 0;
            }
        };

        Iterator iterator = new FilterIterator(stateArray.iterator(), predicate);

        ClustersState state = null;
        while (iterator.hasNext()) {
            state = (ClustersState) iterator.next();
            break;

        }
        return (state != null) ? state.getState() : 0;
    }

    public static ConsumerClusters selectByClusters(String clusters) {
        Predicate predicate = new Predicate() {
            public boolean evaluate(Object object) {
                String id = ((ClustersRelation) object).getId();
                return id.compareTo(clusters) == 0;
            }
        };

        Iterator iterator = new FilterIterator(relationArray.iterator(), predicate);

        ClustersRelation relation = null;
        while (iterator.hasNext()) {
            relation = (ClustersRelation) iterator.next();
            break;
        }

        return (relation != null) ? relation.getClusters() : null;
    }

    public static List<ConsumerClusters> selectByTopic(String topic) {

        List<ConsumerClusters> clusters = new ArrayList<ConsumerClusters>();

        for (int i = 0; i < relationArray.size(); i++) {
            ConcurrentHashMap<String, SubscriptionData> subscriptionTable = relationArray.get(i).getClusters().getSubMap();
            if (subscriptionTable.containsKey(topic)) {
                clusters.add(relationArray.get(i).getClusters());
            }
        }

        return clusters;
    }

    public static void addClusters(String clusters, RemoteChannelData channelinfo) {
        ConsumerClusters manage = selectByClusters(clusters);
        if (manage == null) {
            ConsumerClusters newClusters = new ConsumerClusters(clusters);
            newClusters.attachRemoteChannelData(channelinfo.getClientId(), channelinfo);
            relationArray.add(new ClustersRelation(clusters, newClusters));
        } else if (manage.findRemoteChannelData(channelinfo.getClientId()) != null) {
            manage.detachRemoteChannelData(channelinfo.getClientId());
            manage.attachRemoteChannelData(channelinfo.getClientId(), channelinfo);
        } else {
            String topic = channelinfo.getSubcript().getTopic();
            boolean touchChannel = manage.getSubMap().containsKey(topic);
            if (touchChannel) {
                manage.attachRemoteChannelData(channelinfo.getClientId(), channelinfo);
            } else {
                manage.getSubMap().clear();
                manage.getChannelMap().clear();
                manage.attachRemoteChannelData(channelinfo.getClientId(), channelinfo);
            }
        }
    }

    public static void unLoad(String clientId) {

        for (int i = 0; i < relationArray.size(); i++) {
            String id = relationArray.get(i).getId();
            ConsumerClusters manage = relationArray.get(i).getClusters();

            if (manage.findRemoteChannelData(clientId) != null) {
                manage.detachRemoteChannelData(clientId);
            }

            if (manage.getChannelMap().size() == 0) {
                ClustersRelation relation = new ClustersRelation();
                relation.setId(id);
                relationArray.remove(id);
            }
        }
    }
} public class AckMessageCache extends MessageCache<String> {

    private CyclicBarrier barrier = null;
    private long succTaskCount = 0;

    private AckMessageCache() {

    }

    public long getSuccTaskCount() {
        return succTaskCount;
    }

    private static class AckMessageCacheHolder {

        public static AckMessageCache cache = new AckMessageCache();
    }

    public static AckMessageCache getAckMessageCache() {
        return AckMessageCacheHolder.cache;
    }

    public void parallelDispatch(LinkedList<String> list) {
        List<Callable<Long>> tasks = new ArrayList<Callable<Long>>();
        List<Future<Long>> futureList = new ArrayList<Future<Long>>();
        int startPosition = 0;
        Pair<Integer, Integer> pair = calculateBlocks(list.size(), list.size());
        int numberOfThreads = pair.getRight();
        int blocks = pair.getLeft();

        barrier = new CyclicBarrier(numberOfThreads);

        for (int i = 0; i < numberOfThreads; i++) {
            String[] task = new String[blocks];
            System.arraycopy(list.toArray(), startPosition, task, 0, blocks);
            tasks.add(new AckMessageTask(barrier, task));
            startPosition += blocks;
        }

        ExecutorService executor = Executors.newFixedThreadPool(numberOfThreads);
        try {
            futureList = executor.invokeAll(tasks);
        } catch (InterruptedException ex) {
            Logger.getLogger(AckMessageCache.class.getName()).log(Level.SEVERE, null, ex);
        }

        for (Future<Long> longFuture : futureList) {
            try {
                succTaskCount += longFuture.get();
            } catch (InterruptedException ex) {
                Logger.getLogger(AckMessageCache.class.getName()).log(Level.SEVERE, null, ex);
            } catch (ExecutionException ex) {
                Logger.getLogger(AckMessageCache.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }
} public class AckTaskQueue {

    private static ConcurrentLinkedQueue<ProducerAckMessage> ackQueue = new ConcurrentLinkedQueue<ProducerAckMessage>();

    public static boolean pushAck(ProducerAckMessage ack) {
        return ackQueue.offer(ack);
    }

    public static boolean pushAck(List<ProducerAckMessage> acks) {
        boolean flag = false;
        for (ProducerAckMessage ack : acks) {
            flag = ackQueue.offer(ack);
        }
        return flag;
    }

    public static ProducerAckMessage getAck() {
        return ackQueue.poll();
    }
} public class MessageTaskQueue {

    private static AtomicBoolean isInit = new AtomicBoolean(false);
    private static ConcurrentLinkedQueue<MessageDispatchTask> taskQueue = null;

    private volatile static MessageTaskQueue task = null;

    private MessageTaskQueue() {
    }

    public static MessageTaskQueue getInstance() {
        if (isInit.compareAndSet(false, true)) {
            taskQueue = new ConcurrentLinkedQueue<MessageDispatchTask>();
            task = new MessageTaskQueue();
        }
        return task;
    }

    public boolean pushTask(MessageDispatchTask task) {
        return taskQueue.offer(task);
    }

    public boolean pushTask(List<MessageDispatchTask> tasks) {
        return taskQueue.addAll(tasks);
    }

    public MessageDispatchTask getTask() {
        return taskQueue.poll();
    }
} public class MessageSystemConfig {

    public static final String SystemPropertySocketSndbufSize
            = "com.newlandframework.avatarmq.netty.socket.sndbuf.size";
    public static int SocketSndbufSize
            = Integer.parseInt(System.getProperty(SystemPropertySocketSndbufSize, "65535"));

    public static final String SystemPropertySocketRcvbufSize
            = "com.newlandframework.avatarmq.netty.socket.rcvbuf.size";
    public static int SocketRcvbufSize
            = Integer.parseInt(System.getProperty(SystemPropertySocketRcvbufSize, "65535"));

    public static final String SystemPropertyAckTaskSemaphoreValue
            = "com.newlandframework.avatarmq.semaphore.ackTaskSemaphoreValue";
    public static String AckTaskSemaphoreValue
            = System.getProperty(SystemPropertyAckTaskSemaphoreValue, "Ack");

    public static final String SystemPropertyNotifyTaskSemaphoreValue
            = "com.newlandframework.avatarmq.semaphore.NotifyTaskSemaphoreValue";
    public static String NotifyTaskSemaphoreValue
            = System.getProperty(SystemPropertyNotifyTaskSemaphoreValue, "Notify");

    public static final String SystemPropertySemaphoreCacheHookTimeValue
            = "com.newlandframework.avatarmq.semaphore.hooktime";
    public static int SemaphoreCacheHookTimeValue
            = Integer.parseInt(System.getProperty(SystemPropertySemaphoreCacheHookTimeValue, "5"));

    public static final String SystemPropertyMessageTimeOutValue
            = "com.newlandframework.avatarmq.system.normal.timeout";
    public static int MessageTimeOutValue
            = Integer.parseInt(System.getProperty(SystemPropertyMessageTimeOutValue, "3000"));

    public static final String SystemPropertyAckMessageControllerTimeOutValue
            = "com.newlandframework.avatarmq.system.ack.timeout";
    public static int AckMessageControllerTimeOutValue
            = Integer.parseInt(System.getProperty(SystemPropertyAckMessageControllerTimeOutValue, "1000"));

    public static final String SystemPropertySendMessageControllerPeriodTimeValue
            = "com.newlandframework.avatarmq.system.send.period";
    public static int SendMessageControllerPeriodTimeValue
            = Integer.parseInt(System.getProperty(SystemPropertySendMessageControllerPeriodTimeValue, "3000"));

    public static final String SystemPropertySendMessageControllerTaskCommitValue
            = "com.newlandframework.avatarmq.system.send.taskcommit";
    public static int SendMessageControllerTaskCommitValue
            = Integer.parseInt(System.getProperty(SystemPropertySendMessageControllerTaskCommitValue, "1"));

    public static final String SystemPropertySendMessageControllerTaskSleepTimeValue
            = "com.newlandframework.avatarmq.system.send.sleeptime";
    public static int SendMessageControllerTaskSleepTimeValue
            = Integer.parseInt(System.getProperty(SystemPropertySendMessageControllerTaskSleepTimeValue, "5000"));

    public final static String MessageDelimiter = "@";
    public final static String IpV4AddressDelimiter = ":";
} public class MessageDispatchTask implements Serializable {

    private String clusters;

    private String topic;

    private Message message;

    public String getClusters() {
        return clusters;
    }

    public void setClusters(String clusters) {
        this.clusters = clusters;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public Message getMessage() {
        return message;
    }

    public void setMessage(Message message) {
        this.message = message;
    }

    public boolean equals(Object obj) {
        boolean result = false;
        if (obj != null && MessageDispatchTask.class.isAssignableFrom(obj.getClass())) {
            MessageDispatchTask task = (MessageDispatchTask) obj;
            result = new EqualsBuilder().append(clusters, task.getClusters()).append(topic, task.getTopic()).append(message, task.getMessage())
                    .isEquals();
        }
        return result;
    }
} public class ChannelCache {

    private static ConcurrentHashMap<String, Channel> producerMap
            = new ConcurrentHashMap<String, Channel>();

    public static void pushRequest(String requestId, Channel channel) {
        producerMap.put(requestId, channel);
    }

    public static Channel findChannel(String requestId) {
        Channel channel = producerMap.remove(requestId);
        return channel;
    }
}
import io.netty.channel.Channel;
import java.util.ArrayList;
import java.util.List;
import com.newlandframework.avatarmq.msg.Message;
import com.newlandframework.avatarmq.msg.ProducerAckMessage;
import com.newlandframework.avatarmq.core.SemaphoreCache;
import com.newlandframework.avatarmq.consumer.ConsumerClusters;
import com.newlandframework.avatarmq.consumer.ConsumerContext;
import com.newlandframework.avatarmq.core.AckMessageCache;
import com.newlandframework.avatarmq.core.AckTaskQueue;
import com.newlandframework.avatarmq.core.MessageTaskQueue;
import com.newlandframework.avatarmq.core.MessageSystemConfig;
import com.newlandframework.avatarmq.model.MessageDispatchTask;
import com.newlandframework.avatarmq.core.ChannelCache;
import org.apache.commons.collections.Closure;
import org.apache.commons.collections.Predicate;
import org.apache.commons.collections.ClosureUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.functors.AnyPredicate;
import com.google.common.base.Joiner;
/**
 * Copyright (C) 2016 Newland Group Holding Limited
 *
 * Licensed 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 com.newlandframework.avatarmq.broker;


/**
 * @filename:ProducerMessageHook.java
 * @description:ProducerMessageHook功能模块
 * @author tangjie<https://github.com/tang-jie>
 * @blog http://www.cnblogs.com/jietang/
 * @since 2016-8-11
 */
public class ProducerMessageHook implements ProducerMessageListener {

    private List<ConsumerClusters> clustersSet = new ArrayList<ConsumerClusters>();
    private List<ConsumerClusters> focusTopicGroup = null;

    private void filterByTopic(String topic) {
        Predicate focusAllPredicate = new Predicate() {
            public boolean evaluate(Object object) {
                ConsumerClusters clusters = (ConsumerClusters) object;
                return clusters.findSubscriptionData(topic) != null;
            }
        };

        AnyPredicate any = new AnyPredicate(new Predicate[]{focusAllPredicate});

        Closure joinClosure = new Closure() {
            public void execute(Object input) {
                if (input instanceof ConsumerClusters) {
                    ConsumerClusters clusters = (ConsumerClusters) input;
                    clustersSet.add(clusters);
                }
            }
        };

        Closure ignoreClosure = new Closure() {
            public void execute(Object input) {
            }
        };

        Closure ifClosure = ClosureUtils.ifClosure(any, joinClosure, ignoreClosure);

        CollectionUtils.forAllDo(focusTopicGroup, ifClosure);
    }

    private boolean checkClustersSet(Message msg, String requestId) {