Add RM_RdbLoad and RM_RdbSave module API functions (#11852)

Add `RM_RdbLoad()` and `RM_RdbSave()` to load/save RDB files from the module API. 

In our use case, we have our clustering implementation as a module. As part of this
implementation, the module needs to trigger RDB save operation at specific points.
Also, this module delivers RDB files to other nodes (not using Redis' replication).
When a node receives an RDB file, it should be able to load the RDB. Currently,
there is no module API to save/load RDB files. 


This PR adds four new APIs:
```c
RedisModuleRdbStream *RM_RdbStreamCreateFromFile(const char *filename);
void RM_RdbStreamFree(RedisModuleRdbStream *stream);

int RM_RdbLoad(RedisModuleCtx *ctx, RedisModuleRdbStream *stream, int flags);
int RM_RdbSave(RedisModuleCtx *ctx, RedisModuleRdbStream *stream, int flags);
```

The first step is to create a `RedisModuleRdbStream` object. This PR provides a function to
create RedisModuleRdbStream from the filename. (You can load/save RDB with the filename).
In the future, this API can be extended if needed: 
e.g., `RM_RdbStreamCreateFromFd()`, `RM_RdbStreamCreateFromSocket()` to save/load
RDB from an `fd` or a `socket`. 


Usage:
```c
/* Save RDB */
RedisModuleRdbStream *stream = RedisModule_RdbStreamCreateFromFile("example.rdb");
RedisModule_RdbSave(ctx, stream, 0);
RedisModule_RdbStreamFree(stream);

/* Load RDB */
RedisModuleRdbStream *stream = RedisModule_RdbStreamCreateFromFile("example.rdb");
RedisModule_RdbLoad(ctx, stream, 0);
RedisModule_RdbStreamFree(stream);
```

2 files changed, 164 insertions, 1 deletions

diff --git a/tests/modules/Makefile b/tests/modules/Makefile
index a1f5b074b..d63c8548d 100644
--- a/tests/modules/Makefile
+++ b/tests/modules/Makefile
@@ -61,7 +61,8 @@ TEST_MODULES = \
     publish.so \
     usercall.so \
     postnotifications.so \
-    moduleauthtwo.so
+    moduleauthtwo.so \
+    rdbloadsave.so
 
 .PHONY: all
 
diff --git a/tests/modules/rdbloadsave.c b/tests/modules/rdbloadsave.c
new file mode 100644
index 000000000..687269a5a
--- /dev/null
+++ b/tests/modules/rdbloadsave.c
@@ -0,0 +1,162 @@
+#include "redismodule.h"
+
+#include <stdlib.h>
+#include <unistd.h>
+#include <fcntl.h>
+#include <memory.h>
+#include <errno.h>
+
+/* Sanity tests to verify inputs and return values. */
+int sanity(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    REDISMODULE_NOT_USED(argv);
+    REDISMODULE_NOT_USED(argc);
+
+    RedisModuleRdbStream *s = RedisModule_RdbStreamCreateFromFile("dbnew.rdb");
+
+    /* NULL stream should fail. */
+    if (RedisModule_RdbLoad(ctx, NULL, 0) == REDISMODULE_OK || errno != EINVAL) {
+        RedisModule_ReplyWithError(ctx, strerror(errno));
+        goto out;
+    }
+
+    /* Invalid flags should fail. */
+    if (RedisModule_RdbLoad(ctx, s, 188) == REDISMODULE_OK || errno != EINVAL) {
+        RedisModule_ReplyWithError(ctx, strerror(errno));
+        goto out;
+    }
+
+    /* Missing file should fail. */
+    if (RedisModule_RdbLoad(ctx, s, 0) == REDISMODULE_OK || errno != ENOENT) {
+        RedisModule_ReplyWithError(ctx, strerror(errno));
+        goto out;
+    }
+
+    /* Save RDB file. */
+    if (RedisModule_RdbSave(ctx, s, 0) != REDISMODULE_OK || errno != 0) {
+        RedisModule_ReplyWithError(ctx, strerror(errno));
+        goto out;
+    }
+
+    /* Load the saved RDB file. */
+    if (RedisModule_RdbLoad(ctx, s, 0) != REDISMODULE_OK || errno != 0) {
+        RedisModule_ReplyWithError(ctx, strerror(errno));
+        goto out;
+    }
+
+    RedisModule_ReplyWithSimpleString(ctx, "OK");
+
+ out:
+    RedisModule_RdbStreamFree(s);
+    return REDISMODULE_OK;
+}
+
+int cmd_rdbsave(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    if (argc != 2) {
+        RedisModule_WrongArity(ctx);
+        return REDISMODULE_OK;
+    }
+
+    size_t len;
+    const char *filename = RedisModule_StringPtrLen(argv[1], &len);
+
+    char tmp[len + 1];
+    memcpy(tmp, filename, len);
+    tmp[len] = '\0';
+
+    RedisModuleRdbStream *stream = RedisModule_RdbStreamCreateFromFile(tmp);
+
+    if (RedisModule_RdbSave(ctx, stream, 0) != REDISMODULE_OK || errno != 0) {
+        RedisModule_ReplyWithError(ctx, strerror(errno));
+        goto out;
+    }
+
+    RedisModule_ReplyWithSimpleString(ctx, "OK");
+
+out:
+    RedisModule_RdbStreamFree(stream);
+    return REDISMODULE_OK;
+}
+
+/* Fork before calling RM_RdbSave(). */
+int cmd_rdbsave_fork(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    if (argc != 2) {
+        RedisModule_WrongArity(ctx);
+        return REDISMODULE_OK;
+    }
+
+    size_t len;
+    const char *filename = RedisModule_StringPtrLen(argv[1], &len);
+
+    char tmp[len + 1];
+    memcpy(tmp, filename, len);
+    tmp[len] = '\0';
+
+    int fork_child_pid = RedisModule_Fork(NULL, NULL);
+    if (fork_child_pid < 0) {
+        RedisModule_ReplyWithError(ctx, strerror(errno));
+        return REDISMODULE_OK;
+    } else if (fork_child_pid > 0) {
+        /* parent */
+        RedisModule_ReplyWithSimpleString(ctx, "OK");
+        return REDISMODULE_OK;
+    }
+
+    RedisModuleRdbStream *stream = RedisModule_RdbStreamCreateFromFile(tmp);
+
+    int ret = 0;
+    if (RedisModule_RdbSave(ctx, stream, 0) != REDISMODULE_OK) {
+        ret = errno;
+    }
+    RedisModule_RdbStreamFree(stream);
+
+    RedisModule_ExitFromChild(ret);
+    return REDISMODULE_OK;
+}
+
+int cmd_rdbload(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    if (argc != 2) {
+        RedisModule_WrongArity(ctx);
+        return REDISMODULE_OK;
+    }
+
+    size_t len;
+    const char *filename = RedisModule_StringPtrLen(argv[1], &len);
+
+    char tmp[len + 1];
+    memcpy(tmp, filename, len);
+    tmp[len] = '\0';
+
+    RedisModuleRdbStream *stream = RedisModule_RdbStreamCreateFromFile(tmp);
+
+    if (RedisModule_RdbLoad(ctx, stream, 0) != REDISMODULE_OK || errno != 0) {
+        RedisModule_RdbStreamFree(stream);
+        RedisModule_ReplyWithError(ctx, strerror(errno));
+        return REDISMODULE_OK;
+    }
+
+    RedisModule_RdbStreamFree(stream);
+    RedisModule_ReplyWithSimpleString(ctx, "OK");
+    return REDISMODULE_OK;
+}
+
+int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    REDISMODULE_NOT_USED(argv);
+    REDISMODULE_NOT_USED(argc);
+
+    if (RedisModule_Init(ctx, "rdbloadsave", 1, REDISMODULE_APIVER_1) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
+    if (RedisModule_CreateCommand(ctx, "test.sanity", sanity, "", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
+    if (RedisModule_CreateCommand(ctx, "test.rdbsave", cmd_rdbsave, "", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
+    if (RedisModule_CreateCommand(ctx, "test.rdbsave_fork", cmd_rdbsave_fork, "", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
+    if (RedisModule_CreateCommand(ctx, "test.rdbload", cmd_rdbload, "", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
+    return REDISMODULE_OK;
+}