public void notifyMessage()

in mqtt-ds/src/main/java/org/apache/rocketmq/mqtt/ds/notify/NotifyManager.java [213:248]

36 lines of code
10 McCabe index (conditional complexity)


    public void notifyMessage(Set<MessageEvent> messageEvents) throws
            MQBrokerException, RemotingException, InterruptedException, MQClientException {
        Set<String> connectorNodes = metaPersistManager.getConnectNodeSet();
        if (connectorNodes == null || connectorNodes.isEmpty()) {
            throw new RemotingException("No Connect Nodes");
        }
        for (String node : connectorNodes) {
            boolean result = false;
            try {
                AtomicInteger nodeFailCount = nodeFail.get(node);
                if (nodeFailCount == null) {
                    nodeFailCount = new AtomicInteger();
                    AtomicInteger old = nodeFail.putIfAbsent(node, nodeFailCount);
                    if (old != null) {
                        nodeFailCount = old;
                    }
                }
                if (nodeFailCount.get() > NODE_FAIL_MAX_NUM) {
                    sendEventRetryMsg(messageEvents, 1, node, 0);
                    continue;
                }
                if (result = doNotify(node, messageEvents)) {
                    nodeFailCount.set(0);
                    continue;
                }
                nodeFailCount.incrementAndGet();
            } catch (Exception e) {
                logger.error("", e);
                result = false;
            } finally {
                if (!result) {
                    sendEventRetryMsg(messageEvents, 1, node, 0);
                }
            }
        }
    }