xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

add backend

This commit is contained in:
yihaoDeng 2023-03-22 13:34:19 +00:00
parent b701198dbb
commit 40535b5ede
1 changed files with 5 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
source/libs/stream/src/streamStateRocksdb.c
View File

@ -457,20 +457,20 @@ int32_t streamStatePut_rocksdb(SStreamState* pState, const SWinKey* key, const v
int code = 0; int code = 0;
SStateKey sKey = {.key = *key, .opNum = pState->number}; SStateKey sKey = {.key = *key, .opNum = pState->number};
STREAM_STATE_PUT_ROCKSDB(pState, "default", key, value, vLen); STREAM_STATE_PUT_ROCKSDB(pState, "default", &sKey, value, vLen);
return code; return code;
} }
int32_t streamStateGet_rocksdb(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen) { int32_t streamStateGet_rocksdb(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen) {
int code = 0; int code = 0;
SStateKey sKey = {.key = *key, .opNum = pState->number}; SStateKey sKey = {.key = *key, .opNum = pState->number};
STREAM_STATE_GET_ROCKSDB(pState, "default", key, pVal, pVLen); STREAM_STATE_GET_ROCKSDB(pState, "default", &sKey, pVal, pVLen);
return code; return code;
} }
// todo refactor // todo refactor
int32_t streamStateDel_rocksdb(SStreamState* pState, const SWinKey* key) { int32_t streamStateDel_rocksdb(SStreamState* pState, const SWinKey* key) {
int code = 0; int code = 0;
SStateKey sKey = {.key = *key, .opNum = pState->number}; SStateKey sKey = {.key = *key, .opNum = pState->number};
STREAM_STATE_DEL_ROCKSDB(pState, "default", key); STREAM_STATE_DEL_ROCKSDB(pState, "default", &sKey);
return code; return code;
} }
@ -515,7 +515,7 @@ int32_t streamStateFillDel_rocksdb(SStreamState* pState, const SWinKey* key) {
int32_t streamStateSessionPut_rocksdb(SStreamState* pState, const SSessionKey* key, const void* value, int32_t vLen) { int32_t streamStateSessionPut_rocksdb(SStreamState* pState, const SSessionKey* key, const void* value, int32_t vLen) {
int code = 0; int code = 0;
SStateSessionKey sKey = {.key = *key, .opNum = pState->number}; SStateSessionKey sKey = {.key = *key, .opNum = pState->number};
STREAM_STATE_PUT_ROCKSDB(pState, "sess", key, value, vLen); STREAM_STATE_PUT_ROCKSDB(pState, "sess", &sKey, value, vLen);
if (code == -1) { if (code == -1) {
} }
return code; return code;
@ -948,7 +948,7 @@ int32_t streamStateSessionGet_rocksdb(SStreamState* pState, SSessionKey* key, vo
int32_t streamStateSessionDel_rocksdb(SStreamState* pState, const SSessionKey* key) { int32_t streamStateSessionDel_rocksdb(SStreamState* pState, const SSessionKey* key) {
int code = 0; int code = 0;
SStateSessionKey sKey = {.key = *key, .opNum = pState->number}; SStateSessionKey sKey = {.key = *key, .opNum = pState->number};
STREAM_STATE_DEL_ROCKSDB(pState, "sess", key); STREAM_STATE_DEL_ROCKSDB(pState, "sess", &sKey);
return code; return code;
} }
int32_t streamStateSessionAddIfNotExist_rocksdb(SStreamState* pState, SSessionKey* key, TSKEY gap, void** pVal, int32_t streamStateSessionAddIfNotExist_rocksdb(SStreamState* pState, SSessionKey* key, TSKEY gap, void** pVal,