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

Code before:
static clusterMsgDataPublishMessage *getPublishMessageFirst(clusterMsgDataPublishMulti *msg) {
    clusterMsgDataPublishMessage *initial = (clusterMsgDataPublishMessage *)&(msg->bulk_data);
    return initial;
}

static clusterMsgDataPublishMessage *getPublishMessageNext(clusterMsgDataPublishMessage *message_cursor) {
    clusterMsgDataPublishMessage *next =
        (clusterMsgDataPublishMessage *)((char *)message_cursor->message_data + ntohl(message_cursor->message_len));
    return next;
}

void writePublishMessageToCursor(clusterMsgDataPublishMessage *cursor, robj *data) {
    uint32_t data_len = sdslen(data->ptr);
    cursor->message_len = htonl(data_len);
    memcpy(cursor->message_data, data->ptr, data_len);
    return;
}

static robj *readPublishMessageFromCursor(clusterMsgDataPublishMessage *cursor) {
    uint32_t data_len;
    data_len = ntohl(cursor->message_len);
    robj *data = (createStringObject((char *)cursor->message_data, data_len));
    return data;
}

static uint32_t getPublishMsgLength(clusterMsgDataPublishMessage *cursor) {
    uint32_t msg_length = ntohl(cursor->message_len);
    return msg_length;
}

int pubsubProcessLightPacket(clusterLink *link, uint16_t type) {
    clusterMsgLight *hdr = (clusterMsgLight *)link->rcvbuf;
    robj *channel, *message;
    uint64_t data_count;

    /* Don't bother creating useless objects if there are no
     * Pub/Sub subscribers. */
    if ((type == CLUSTERMSG_TYPE_PUBLISH_LIGHT && serverPubsubSubscriptionCount() > 0) ||
        (type == CLUSTERMSG_TYPE_PUBLISHSHARD_LIGHT && serverPubsubShardSubscriptionCount() > 0)) {
        data_count = ntohu64(hdr->data.publish.msg.data_count);
        clusterMsgDataPublishMessage *cursor = getPublishMessageFirst(&hdr->data.publish.msg);
        channel = readPublishMessageFromCursor(cursor);
        while (--data_count) {
            cursor = getPublishMessageNext(cursor);
            message = readPublishMessageFromCursor(cursor);
            pubsubPublishMessage(channel, message, type == CLUSTERMSG_TYPE_PUBLISHSHARD_LIGHT);
            decrRefCount(message);
        }
        decrRefCount(channel);
    }
    return 1;

Review comment (suggestion):
To be symmetrical with other methods:


```suggestion
static uint32_t getPublishMessageLength(clusterMsgDataPublishMessage *cursor) {
```

Code after:
            /* Unknown type, we will ignore it but log what happened. */
            serverLog(LL_WARNING, "Received unknown extension type %d", type);
        }

        /* We know this will be valid since we validated it ahead of time */
        ext = getNextPingExt(ext);
    }

    /* If the node did not send us a hostname extension, assume
     * they don't have an announced hostname. Otherwise, we'll
     * set it now. */
    updateAnnouncedHostname(sender, ext_hostname);
    updateAnnouncedHumanNodename(sender, ext_humannodename);
    updateAnnouncedClientIpV4(sender, ext_clientipv4);
    updateAnnouncedClientIpV6(sender, ext_clientipv6);
    /* If the node did not send us a shard-id extension, it means the sender
     * does not support it (old version), node->shard_id is randomly generated.
     * A cluster-wide consensus for the node's shard_id is not necessary.
     * The key is maintaining consistency of the shard_id on each individual 7.2 node.
     * As the cluster progressively upgrades to version 7.2, we can expect the shard_ids
     * across all nodes to naturally converge and align.
     *
     * If sender is a replica, set the shard_id to the shard_id of its primary.
     * Otherwise, we'll set it now. */
    if (ext_shardid == NULL) ext_shardid = clusterNodeGetPrimary(sender)->shard_id;

    updateShardId(sender, ext_shardid);
}

static clusterNode *getNodeFromLinkAndMsg(clusterLink *link, clusterMsg *hdr) {
    clusterNode *sender;
    if (link->node && !nodeInHandshake(link->node)) {
        /* If the link has an associated node, use that so that we don't have to look it
         * up every time, except when the node is still in handshake, the node still has
         * a random name thus not truly "known". */
        sender = link->node;
    } else {
        /* Otherwise, fetch sender based on the message */
        sender = clusterLookupNode(hdr->sender, CLUSTER_NAMELEN);
        /* We know the sender node but haven't associate it with the link. This must
         * be an inbound link because only for inbound links we didn't know which node
         * to associate when they were created. */
        if (sender && !link->node) {
            setClusterNodeToInboundClusterLink(sender, link);
        }
    }
    return sender;
}

static void clusterProcessPublishPacket(clusterMsgDataPublish *publish_data, uint16_t type) {
    robj *channel, *message;