public class ClientChannelInfo {
    private final Channel channel; //关联的channel;
    private final String clientId; //客户端的Id;
    private SubscriptionInfo subcript; //该客户端的订阅信息.

    public ClientChannelInfo(Channel channel, String clientId) {
        this.channel = channel;
        this.clientId = clientId;
    }

    public Channel getChannel() {
        return channel;
    }

    public String getClientId() {
        return clientId;
    }

    public SubscriptionInfo getSubcript() {
        return subcript;
    }

    public void setSubcript(SubscriptionInfo subcript) {
        this.subcript = subcript;
    }
} public class ConsumerManager {
    //存储groupid -> GroupInfo
     private static  final ConcurrentHashMap<String/*Group*/,ConsumerGroupInfo> consumerTable = new ConcurrentHashMap<String, ConsumerGroupInfo>(1024);
     //存储每个group状态的表
    /**
     * 0:正常
     * -1:连接断开
     * 1:标识超时
     */
    private static final ConcurrentHashMap<String/*Group*/,Integer> consumerStat = new ConcurrentHashMap<String, Integer>(1024);

    public static void putConsumerGroupStat(String group,Integer stat){
        consumerStat.put(group,stat);
    }
    public static Integer getConsumerGroupStat(String group){
        return consumerStat.get(group);
    }
    public static ConsumerGroupInfo findGroupByGroupID(final String groupId){
        return consumerTable.get(groupId);
    }

    /**
     * 按照groupid 和topic寻找
     * @param groupId
     * @param topic
     * @return
     */
    public static ConsumerGroupInfo findGroupByGroupID(final String groupId,final String topic){
        ConsumerGroupInfo groupinfo = findGroupByGroupID(groupId);
        if(groupinfo.getSubscriptionTable().containsKey(topic)){
            return groupinfo;
        }
        return null;
    }

    /**
     * 查找所有订阅topic主题的消费组的信息
     * @param topic
     * @return
     */
    public static List<ConsumerGroupInfo> findGroupByTopic(final String topic){
        List<ConsumerGroupInfo> groups = new ArrayList<ConsumerGroupInfo>();
        for(Map.Entry<String,ConsumerGroupInfo> entry:consumerTable.entrySet()){
            if(entry.getValue().getSubscriptionTable().containsKey(topic)){
                groups.add(entry.getValue());
            }
        }
        return groups;
    }

    /**
     *　查找订阅topic且过滤条件一致的组
     * @param topic
     * @param property
     * @param value
     * @return
     */
    public static List<ConsumerGroupInfo> findGroupByTopic(final  String topic,final  String property,final String value ){
        List<ConsumerGroupInfo> groups = new ArrayList<ConsumerGroupInfo>();
        for(Map.Entry<String,ConsumerGroupInfo> entry:consumerTable.entrySet()) {
            ConcurrentHashMap<String,SubscriptionInfo> subscriptionTable = entry.getValue().getSubscriptionTable();
          //这个group订阅了topic并且过滤条件一致就返回
            if(subscriptionTable.containsKey(topic)){
                   //判断过滤条件是否一致，当过滤条件为空的时候 就是符合的
                if(subscriptionTable.get(topic).getFitlerName() == property && subscriptionTable.get(topic).getFitlerValue() == value){
                      groups.add(entry.getValue());
                    //如果该group 的过滤条件为空，则认为也是符合匹配的
                  }else if(subscriptionTable.get(topic).getFitlerName().isEmpty()&& subscriptionTable.get(topic).getFitlerValue().isEmpty()){
                      groups.add(entry.getValue());
                  }
            }
        }
        return groups;
    }

    /**
     * 根据clientID查找所属的消费组信息
     * @param clientID
     * @return
     */
    public static ConsumerGroupInfo findGroupByClientID(final String clientID){
        for(Map.Entry<String,ConsumerGroupInfo> entry:consumerTable.entrySet()) {
            if(entry.getValue().findChannel(clientID) != null){
                return entry.getValue();
            }

        }
              return null;
    }

    /**
     * 根据group和client 找到client对应的ClientChannelInfo;
     * @param group
     * @param clientId
     * @return
     */
    public static ClientChannelInfo findChannelInfoByID(final String group,final String clientId){
        for(Map.Entry<String,ConsumerGroupInfo> entry:consumerTable.entrySet()){
            ConsumerGroupInfo consumerGroupInfo = consumerTable.get(group);
            if(consumerGroupInfo != null){
                return consumerGroupInfo.findChannel(clientId);
            }
        }
        return null;
    }

    /**
     * 添加一个channelinfo 进入本组,如果组不存在,则创建.
     * @param group
     * @param channelInfo
     */
    public static void addGroupInfo(String group,ClientChannelInfo channelInfo){
          //如果此组不存在

       // System.out.println("#####"+findGroupByGroupID(group)==null?"yang":"long");
           if(findGroupByGroupID(group) == null) {
            //   System.out.println("addGroupInfo0");
                ConsumerGroupInfo newgroup = new ConsumerGroupInfo(group);
               //把此消费者添加到新创建的消费组里.
              // System.out.println("addGroupInfo1");
                newgroup.addChannel(channelInfo.getClientId(),channelInfo);
               //增加新的订阅信息.
              // System.out.println("addGroupInfo2");
               newgroup.addSubscript(channelInfo.getSubcript());
              // System.out.println("addGroupInfo3");

               consumerTable.put(group,newgroup); //加入一个新的组
              // System.out.println("addGroupInfo4");
               System.out.println("create group: "+group+ " topic:"+channelInfo.getSubcript().getTopic());
           }else if(findGroupByGroupID(group).findChannel(channelInfo.getClientId()) != null){
               //之前存在这个clientId,说明是重新连上的
               //把之前关联的旧的channel删除
               findGroupByGroupID(group).removeChannel(channelInfo.getClientId());
               findGroupByGroupID(group).addChannel(channelInfo.getClientId(),channelInfo);
               System.out.println("consumer reconnected");

           }else{
               //这里处理的是当消费者向broker发送Stop类型的消息时
               try{
                  // System.out.println("else");
                   String filterName = channelInfo.getSubcript().getFitlerName();
                   String topic = channelInfo.getSubcript().getTopic();

                   ConsumerGroupInfo groupInfo = findGroupByGroupID(group);
                   if(groupInfo.getSubscriptionTable().containsKey(topic)){
                       SubscriptionInfo sub = groupInfo.findSubscriptionData(topic);
                       if(filterName == null){
                           if(sub.getFitlerName() ==  null){
                               //添加进组
                               findGroupByGroupID(group).addChannel(channelInfo.getClientId(),channelInfo);
                               System.out.println("add into group 1");
                           }else{
                               //清除之前的组员,再添加进入
                               System.out.println("update subscript 1");
                               findGroupByGroupID(group).getSubscriptionTable().clear();
                               findGroupByGroupID(group).addSubscript(channelInfo.getSubcript());
                               findGroupByGroupID(group).getChannelInfoTable().clear();
                               findGroupByGroupID(group).addChannel(channelInfo.getClientId(),channelInfo);
                           }
                       }else {
                           if(sub.getFitlerName() != null && sub.getFitlerName().equals(filterName)){
                               //添加进入组
                               findGroupByGroupID(group).addChannel(channelInfo.getClientId(),channelInfo);
                               System.out.println("add into group 2");
                           }else{
                               //清除之前的组员,在添加进入
                               System.out.println("update subscript 2");
                               //如果加入的是新的订阅,之前的就无效了
                               findGroupByGroupID(group).getSubscriptionTable().clear();
                               findGroupByGroupID(group).addSubscript(channelInfo.getSubcript());

                               findGroupByGroupID(group).getChannelInfoTable().clear();
                               findGroupByGroupID(group).addChannel(channelInfo.getClientId(), channelInfo);
                           }
                       }
                   }else {
                       //清除之前的组员,在添加进入
                       System.out.println("update subscript 3");
                       //如果加入的是新的订阅,之前的就无效了.
                       findGroupByGroupID(group).getSubscriptionTable().clear();
                       findGroupByGroupID(group).addSubscript(channelInfo.getSubcript());

                       findGroupByGroupID(group).getChannelInfoTable().clear();
                       findGroupByGroupID(group).addChannel(channelInfo.getClientId(), channelInfo);
                   }
               }catch (Exception e){
                   e.printStackTrace();
               }
           }
    }
    //获取某个组某个主题的订阅信息

    /**
     * 获取某个组某个主题的订阅信息
     * @param group
     * @param topic
     * @return
     */
    public static SubscriptionInfo findSubscriptionInfo(final String group,final String topic){
        ConsumerGroupInfo consumerGroupInfo = findGroupByGroupID(group);
        if(consumerGroupInfo != null){
            return consumerGroupInfo.findSubscriptionData(topic);
        }
        return  null;
    }

    /**
     *  消费者在断开连接时,移除对应的订阅.
     * @param channel
     */
    public static  void  consumerDisconnect(Channel channel){
        for(Map.Entry<String,ConsumerGroupInfo> entry:consumerTable.entrySet()){
            entry.getValue().removeChannel(channel);
        }
    }

    /**
     * 根据channel ,移除相应的信息,如果channelInfoTable为空后,就删除组.
     * @param clientId
     */
    public static void stopConsumer(String clientId){
        for(Map.Entry<String,ConsumerGroupInfo> entry:consumerTable.entrySet()){
            String groupId = entry.getKey();
            ConsumerGroupInfo info = entry.getValue();
            if(info.findChannel(clientId) !=null){
                info.removeChannel(clientId);
                System.out.println(clientId+" stop");
            }
            if(info.getChannelInfoTable().isEmpty()){
                consumerTable.remove(groupId);
                System.out.println("remove group: "+ groupId);
            }
        }
    }

} public class SubscriptionInfo {
    private String topic; //主题
    private String subString; //
    private String fitlerName; //属性过滤名
    private String fitlerValue; //属性过滤值


    public String getTopic() {
        return topic;
    }

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

    public String getSubString() {
        return subString;
    }

    public void setSubString(String subString) {
        this.subString = subString;
    }

    public String getFitlerName() {
        return fitlerName;
    }

    public void setFitlerName(String fitlerName) {
        this.fitlerName = fitlerName;
    }

    public String getFitlerValue() {
        return fitlerValue;
    }

    public void setFitlerValue(String fitlerValue) {
        this.fitlerValue = fitlerValue;
    }
} public class TaskManager {
    private static ConcurrentLinkedQueue<SendTask> taskQueue = new ConcurrentLinkedQueue<SendTask>();
    //需要重新发送的队列
    private static ConcurrentLinkedQueue<SendTask> resendTaskQueue = new ConcurrentLinkedQueue<SendTask>();
    //保存重发的任务
    private static Map<String, SendTask> map = new HashMap<String, SendTask>();

    /**
     * 添加一个发送任务
     *
     * @param task
     */
    public static boolean pushTask(SendTask task) {
        return taskQueue.offer(task);
    }

    public static boolean pushTask(List<SendTask> tasks) {
        boolean flag = false;
        for (SendTask sendTask : tasks) {
            flag = taskQueue.offer(sendTask);
        }
        return flag;
    }

    public static SendTask getTask() {
        return taskQueue.poll();
    }

    /**
     * 恢复之前的发送任务
     */

    public static void RecoverySendTask() {
        if (FlushTool.log != null) {
            try {
                //从文件中读出未发送的任务
                List<SendTask> list = FlushTool.log.Restore();
                System.out.println("recover size: " + list.size());
                //从文件读取任务,然后增加其SendTask.
                for (SendTask sendTask : list) {
                    pushTask(sendTask);
                    SemaphoreManager.increase("SendTask");
                }
            } catch (ClassNotFoundException e) {
                e.printStackTrace();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    public static boolean pushResendTask(SendTask task){
        String key = task.getGroupId()+task.getTopic()+task.getMessage().getMsgId();
        map.put(key,task);
        return resendTaskQueue.offer(task);
    }
    public static SendTask getResendTask(){
        //出队操作
        SendTask task = resendTaskQueue.poll();
        if(task == null){
            return null;
        }
        String key = task.getGroupId()+task.getTopic()+task.getMessage().getMsgId();
        map.remove(key);
        return task;
    }
    public static int getResendNumber(){
        return resendTaskQueue.size();
    }
    public static boolean findInResend(String groupID,String Topic,String MsgId){
        String key = groupID + Topic + MsgId;
        return map.containsKey(key);
    }
} public class LogTask {
    private SendTask task;
    /**
     * status: 0代表message 已经被接收到了
     * status: 1代表消息已经成功的发送给了客户端
     */
    private int status;

    public LogTask(SendTask task,int status){
        this.task = task;
        this.status = status;
    }

    public SendTask getTask() {
        return task;
    }

    public void setTask(SendTask task) {
        this.task = task;
    }

    public int getStatus() {
        return status;
    }

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

    @Override
    public String toString() {
        return "LogTask{" +
                "task=" + task +
                ", status=" + status +
                '}';
    }
} public class SendTask implements Serializable {
    private String groupId; //组id
    private String topic; //主题
    private Message message;

    public String getGroupId() {
        return groupId;
    }

    public void setGroupId(String groupId) {
        this.groupId = groupId;
    }

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

    @Override
    public boolean equals(Object obj) {
        if(this == obj)
            return true;
        if(obj == null)
            return false;
        if(getClass() != obj.getClass())
            return false;
        SendTask other= (SendTask)obj;
        if(groupId == null){
            if(other.getGroupId() != null)
                return false;
        }else {
            if(other.getGroupId() == null)
                return false;
            else if(!groupId.equals(other.getGroupId()))
                return false;
        }
        if(topic == null){
            if(other.getTopic() != null)
                return false;
        }else {
            if(other.getTopic() == null)
                return false;
            else if (!topic.equals(other.getTopic()))
                return false;
        }
        if(message == null){
            if(other.getMessage() != null)
                return false;
        }else{
            if(other.getMessage() == null)
                return false;
            else if(message.equals(other.getMessage()))
                return false;
        }
        return true;
    }

} public class SubscriptRequestinfo {
    private String groupId; //消费者属于哪个消费组.
    private String topic; //消费者要订阅的主题.
    private String propertieName; //订阅的过滤属性名
    private String propertieValue; //订阅的过滤值
    private String clientKey; //客户端的id.

    public String getGroupId() {
        return groupId;
    }

    public void setGroupId(String groupId) {
        this.groupId = groupId;
    }

    public String getTopic() {
        return topic;
    }

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

    public String getPropertieName() {
        return propertieName;
    }

    public void setPropertieName(String propertieName) {
        this.propertieName = propertieName;
    }

    public String getPropertieValue() {
        return propertieValue;
    }

    public void setPropertieValue(String propertieValue) {
        this.propertieValue = propertieValue;
    }

    public String getClientKey() {
        return clientKey;
    }

    public void setClientKey(String clientKey) {
        this.clientKey = clientKey;
    }
} public class ConsumeResult {
    private ConsumeStatus status = ConsumeStatus.FAIL;
    private String info;
    private String groupID;
    private String topic;
    private String msgId;

    public ConsumeStatus getStatus() {
        return status;
    }

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

    public String getInfo() {
        return info;
    }

    public void setInfo(String info) {
        this.info = info;
    }

    public String getGroupID() {
        return groupID;
    }

    public void setGroupID(String groupID) {
        this.groupID = groupID;
    }

    public String getTopic() {
        return topic;
    }

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

    public String getMsgId() {
        return msgId;
    }

    public void setMsgId(String msgId) {
        this.msgId = msgId;
    }
} public enum  ConsumeStatus {
    SUCCESS,
    FAIL;
} public class Message implements Serializable{
    //private static final long serialVersionUID = 5295808332504208830L;
    private String topic;
    private byte[] body;
    private String msgId;
    private long bornTime;
    private Map<String,String> properties = new HashMap<String, String>();

    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 String getMsgId() {
        return msgId;
    }

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

    public long getBornTime() {
        return bornTime;
    }

    public void setBornTime(long bornTime) {
        this.bornTime = bornTime;
    }

    public Map<String, String> getProperties() {
        return properties;
    }

    public void setProperties(Map<String, String> properties) {
        this.properties = properties;
    }
    public String getProperty(String key){
        return (String)this.properties.get(key);
    }
    public void setProperty(String key,String value){
        this.properties.put(key,value);
    }

    @Override
    public boolean equals(Object obj) {
        if(this == obj)
            return true;
        if(obj == null)
            return  false;
        if(getClass() != obj.getClass())
            return false;
        Message other=(Message)obj;
        if(topic == null){
            if(other.getTopic() != null)
                return false;
        }else{
            if(other.getTopic() == null)
                return  false;
            else if(!other.getTopic().equals(topic))
                return  false;
        }
        if(body == null){
            if(other.getBody() != null)
                return false;
        }else{
            if(other.getBody() == null)
                return  false;
            if(body.length != other.getBody().length)
                return  false;
            else
                for(int i = 0;i < body.length; ++i){
                    if(body[i] != other.getBody()[i])
                        return false;
                }
        }

        if(properties==null)
        {
            if(other.getProperties()!=null)
                return false;
        }
        else
        {
            if(other.getProperties()==null)
                return false;
            else if(!properties.equals(other.getProperties()))
                return false;
        }

        return true;
    }

    @Override
    public int hashCode() {
        int result = topic.hashCode();
        result += properties.hashCode();
        if(body != null)
            for(int i = 0; i < body.length; ++i){
                result += body[i];
            }
        return  result;
    }
} public class Tool {
    private static Map<Class<?>,Schema<?>> cachedSchema = new ConcurrentHashMap<Class<?>, Schema<?>>();
    private static Objenesis objenesis = new ObjenesisStd(true);
    private Tool(){

    }
    @SuppressWarnings("unchecked")
    private static <T> Schema<T>  getSchema(Class<T> cls){
        Schema<T> schema = (Schema<T>) cachedSchema.get(cls);
        if(schema == null){
            schema = RuntimeSchema.createFrom(cls);
            if(schema != null){
                cachedSchema.put(cls,schema);
            }
        }
        return  schema;
    }
    /**
     * 序列化(对象 --> 字节数组)
     */
    @SuppressWarnings("uncheched")
    public static <T> byte[] serialize(T obj){
        Class<T> cls = (Class<T>)obj.getClass();
        LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
        try {
            Schema<T> schema = getSchema(cls);
            return ProtostuffIOUtil.toByteArray(obj,schema,buffer);
        }catch (Exception e){
            throw new IllegalStateException(e.getMessage(),e);
        }finally {
            buffer.clear();
        }
    }
    /**
     * 反序列化:字节数组 --->对象
     */
    public static <T> T deserialize(byte[] data,Class<T> cls){
        try {
            T message = (T)objenesis.newInstance(cls);
            Schema<T> schema = getSchema(cls);
            ProtostuffIOUtil.mergeFrom(data,message,schema);
            return message;
        }catch (Exception e){
            throw new IllegalStateException(e.getMessage(),e);
        }
    }
}
import broker.ClientChannelInfo;
import broker.ConsumerManager;
import broker.SubscriptionInfo;
import broker.TaskManager;
import file.LogTask;
import model.SendTask;
import model.SubscriptRequestinfo;
import smq.ConsumeResult;
import smq.ConsumeStatus;
import smq.Message;
import tool.Tool;
package broker.netty;


/**
 * Created by yang on 16-11-25.
 */
public class ConsumerMessageListener extends MessageListener {
    @Override
    void onConsumerResultReceived(ConsumeResult msg) {
        if(msg.getStatus() == ConsumeStatus.SUCCESS) { //消费成功的话,找到对应的队列删除那个消息,然后继续发送下一个消息.