Repository: valkey-io/valkey
File: src/cluster_legacy.c (C)

Code before:
}

/* This function gets a cluster node ID string as target, the same way the nodes
 * addresses are represented in the modules side, resolves the node, and sends
 * the message. If the target is NULL the message is broadcasted.
 *
 * The function returns C_OK if the target is valid, otherwise C_ERR is
 * returned. */
int clusterSendModuleMessageToTarget(const char *target,
                                     uint64_t module_id,
                                     uint8_t type,
                                     const char *payload,
                                     uint32_t len) {
    clusterNode *node = NULL;

    if (target != NULL) {
        node = clusterLookupNode(target, strlen(target));
        if (node == NULL || node->link == NULL) return C_ERR;
    }

    clusterSendModule(target ? node->link : NULL, module_id, type, payload, len);
    return C_OK;
}

/* -----------------------------------------------------------------------------
 * CLUSTER Pub/Sub support
 *
 * If `sharded` is 0:
 * For now we do very little, just propagating [S]PUBLISH messages across the whole
 * cluster. In the future we'll try to get smarter and avoiding propagating those
 * messages to hosts without receives for a given channel.
 * Otherwise:
 * Publish this message across the slot (primary/replica).
 * -------------------------------------------------------------------------- */
void clusterPropagatePublish(robj *channel, robj **messages, int count, int sharded) {
    clusterMsgSendBlock *msgblock, *msgblock_light;
    int i;
    msgblock_light = clusterCreatePublishLightMsgBlock(
        channel, messages, count, sharded ? CLUSTERMSG_TYPE_PUBLISHSHARD_LIGHT : CLUSTERMSG_TYPE_PUBLISH_LIGHT);
    if (!sharded) {
        clusterBroadcastMessage(msgblock_light);
        clusterMsgSendBlockDecrRefCount(msgblock_light);
        if (!server.cluster->is_light_hdr_supported) {
            for (i = 0; i < count; i++) {
                msgblock = clusterCreatePublishMsgBlock(channel, messages[i], CLUSTERMSG_TYPE_PUBLISH);
                clusterBroadcastMessage(msgblock);
                clusterMsgSendBlockDecrRefCount(msgblock);
            }
        }
        return;
    }

Review comment (suggestion):
Like the idea of maintaining `is_light_hdr_supported` across cluster. 

Do we use that to generate only one type of msg block for each propagation? Once the entire cluster becomes homogenous we send out light hdr message for all. I'm fine with either, would prefer readability over gain for a temporal state.

Code after:
                                     uint8_t type,
                                     const char *payload,
                                     uint32_t len) {
    clusterNode *node = NULL;

    if (target != NULL) {
        node = clusterLookupNode(target, strlen(target));
        if (node == NULL || node->link == NULL) return C_ERR;
    }

    clusterSendModule(target ? node->link : NULL, module_id, type, payload, len);
    return C_OK;
}

/* -----------------------------------------------------------------------------
 * CLUSTER Pub/Sub support
 *
 * If `sharded` is 0:
 * For now we do very little, just propagating [S]PUBLISH messages across the whole
 * cluster. In the future we'll try to get smarter and avoiding propagating those
 * messages to hosts without receives for a given channel.
 * Otherwise:
 * Publish this message across the slot (primary/replica).
 * -------------------------------------------------------------------------- */
void clusterPropagatePublish(robj *channel, robj *message, int sharded) {
    clusterMsgSendBlock *msgblock, *msgblock_light;
    msgblock_light = clusterCreatePublishMsgBlock(channel, message, 1, sharded);
    /* We will only create msgblock with normal hdr if there are any nodes that do not support light hdr */
    msgblock = NULL;
    ClusterNodeIterator iter;
    if (sharded) {
        clusterNodeIterInitMyShard(&iter);
    } else {
        clusterNodeIterInitAllNodes(&iter);
    }

    clusterNode *node;
    while ((node = clusterNodeIterNext(&iter)) != NULL) {
        if (node->flags & (CLUSTER_NODE_MYSELF | CLUSTER_NODE_HANDSHAKE)) continue;
        if (nodeSupportsLightMsgHdr(node)) {
            clusterSendMessage(node->link, msgblock_light);
        } else {
            if (msgblock == NULL) {
                msgblock = clusterCreatePublishMsgBlock(channel, message, 0, sharded);
            }
            clusterSendMessage(node->link, msgblock);
        }
    }
    clusterNodeIterReset(&iter);
    if (msgblock != NULL) clusterMsgSendBlockDecrRefCount(msgblock);
    clusterMsgSendBlockDecrRefCount(msgblock_light);