@Override
    public void run() {
        log.info("SubscribeThread started");
        while ((!this.isInterrupted()) && (!markToBeRemoved)) {
            try {
                cache.drainTo(messages);

                // wait for message
                RoutingMessage routingMessage = messages.poll(200, TimeUnit.MILLISECONDS);
                if (routingMessage != null) {

                    // need this list in case one subscriber has more than one
                    // tag that match with the routingMessage.getTags()
                    // es: "pizza","pasta" and the tag is "pizza" the message must be send only one time.
                    ArrayList<T> tmpAlreadySent = new ArrayList<>();
                    routingMessage.getTags().forEach(tag -> {
                        // for each tag it will find the subscribersID to the indexMap
                        ArrayList<T> subList = tagIndex.getSubscriberIdsByTag(tag);
                        if (subList != null) {
                            for (T subscribeID : subList) {
                                Subscriber<T> sub = this.subscribers.get(subscribeID);
                                if (tmpAlreadySent.indexOf(subscribeID) < 0) {
                                    // if the message has not dispatched to the consumer with subscribe-id = subscribeID
                                    sub.getMessageHandlers().forEach(x ->
                                                    resources.getWorkerServiceThread().submit(() -> {
                                                                try {
                                                                    synchronized (x) { // this synchronized guarantees one message at time for handle
                                                                        x.handleMessage(routingMessage.getHeader(), routingMessage.getBody(), tag,subscribeID);
                                                                    }
                                                                    workerDelivered.inc();
                                                                } catch (Exception e) {
                                                                    log.severe("error handlerMessage:" + e);
                                                                    // must add a error handler
                                                                }
                                                            }
                                                    )
                                    );
                                    tmpAlreadySent.add(subscribeID);
                                }
                            }
                        } /// else that means the tag has not subscriber

                    });
                    tmpAlreadySent.clear();
                    messagesDelivered.inc();
                }


            } catch (InterruptedException e) {
                if (!this.isAlive())
                    break;

            }
        }
    }