mirror of
http://github.com/valkey-io/valkey
synced 2024-11-22 18:54:58 +00:00
8ad8f0f9d8
When a connection that's subscribe to a channel emits PUBLISH inside MULTI-EXEC,
the push notification messes up the EXEC response.
e.g. MULTI, PING, PUSH foo bar, PING, EXEC
the EXEC's response will contain: PONG, {message foo bar}, 1. and the second PONG
will be delivered outside the EXEC's response.
Additionally, this PR changes the order of responses in case of a plain PUBLISH (when
the current client also subscribed to it), by delivering the push after the command's
response instead of before it.
This also affects modules calling RM_PublishMessage in a similar way, so that we don't
run the risk of getting that push mixed together with the module command's response.
58 lines
1.8 KiB
C
58 lines
1.8 KiB
C
#include "redismodule.h"
|
|
#include <string.h>
|
|
#include <assert.h>
|
|
#include <unistd.h>
|
|
|
|
#define UNUSED(V) ((void) V)
|
|
|
|
int cmd_publish_classic_multi(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
|
|
{
|
|
if (argc < 3)
|
|
return RedisModule_WrongArity(ctx);
|
|
RedisModule_ReplyWithArray(ctx, argc-2);
|
|
for (int i = 2; i < argc; i++) {
|
|
int receivers = RedisModule_PublishMessage(ctx, argv[1], argv[i]);
|
|
RedisModule_ReplyWithLongLong(ctx, receivers);
|
|
}
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
int cmd_publish_classic(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
|
|
{
|
|
if (argc != 3)
|
|
return RedisModule_WrongArity(ctx);
|
|
|
|
int receivers = RedisModule_PublishMessage(ctx, argv[1], argv[2]);
|
|
RedisModule_ReplyWithLongLong(ctx, receivers);
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
int cmd_publish_shard(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
|
|
{
|
|
if (argc != 3)
|
|
return RedisModule_WrongArity(ctx);
|
|
|
|
int receivers = RedisModule_PublishMessageShard(ctx, argv[1], argv[2]);
|
|
RedisModule_ReplyWithLongLong(ctx, receivers);
|
|
return REDISMODULE_OK;
|
|
}
|
|
|
|
int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
|
|
UNUSED(argv);
|
|
UNUSED(argc);
|
|
|
|
if (RedisModule_Init(ctx,"publish",1,REDISMODULE_APIVER_1)== REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx,"publish.classic",cmd_publish_classic,"",0,0,0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx,"publish.classic_multi",cmd_publish_classic_multi,"",0,0,0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
if (RedisModule_CreateCommand(ctx,"publish.shard",cmd_publish_shard,"",0,0,0) == REDISMODULE_ERR)
|
|
return REDISMODULE_ERR;
|
|
|
|
return REDISMODULE_OK;
|
|
}
|