mirror of
https://github.com/logos-messaging/nim-sds.git
synced 2026-01-04 07:03:09 +00:00
initial addition of dependencies request
This commit is contained in:
parent
79f42fb461
commit
c0431d53a6
@ -30,6 +30,14 @@ proc allocSharedSeq*[T](s: seq[T]): SharedSeq[T] =
|
|||||||
return (cast[ptr UncheckedArray[T]](data), s.len)
|
return (cast[ptr UncheckedArray[T]](data), s.len)
|
||||||
|
|
||||||
proc deallocSharedSeq*[T](s: var SharedSeq[T]) =
|
proc deallocSharedSeq*[T](s: var SharedSeq[T]) =
|
||||||
|
if not s.data.isNil:
|
||||||
|
when T is cstring:
|
||||||
|
# For array of cstrings, deallocate each string first
|
||||||
|
for i in 0 ..< s.len:
|
||||||
|
if not s.data[i].isNil:
|
||||||
|
# Deallocate each cstring
|
||||||
|
deallocShared(s.data[i])
|
||||||
|
|
||||||
deallocShared(s.data)
|
deallocShared(s.data)
|
||||||
s.len = 0
|
s.len = 0
|
||||||
|
|
||||||
@ -48,6 +56,18 @@ proc allocSharedSeqFromCArray*[T](arr: ptr T, len: int): SharedSeq[T] =
|
|||||||
if arr.isNil or len <= 0:
|
if arr.isNil or len <= 0:
|
||||||
return (nil, 0)
|
return (nil, 0)
|
||||||
|
|
||||||
let data = allocShared(sizeof(T) * len)
|
when T is cstring:
|
||||||
copyMem(data, arr, sizeof(T) * len)
|
# Special handling for arrays of cstrings
|
||||||
return (cast[ptr UncheckedArray[T]](data), len)
|
let data = cast[ptr UncheckedArray[cstring]](allocShared(sizeof(cstring) * len))
|
||||||
|
let cstrArr = cast[ptr UncheckedArray[cstring]](arr)
|
||||||
|
|
||||||
|
for i in 0 ..< len:
|
||||||
|
# Use the existing alloc proc to properly allocate each cstring
|
||||||
|
data[i] = cstrArr[i].alloc()
|
||||||
|
|
||||||
|
return (data, len)
|
||||||
|
else:
|
||||||
|
# Original handling for non-cstring types
|
||||||
|
let data = allocShared(sizeof(T) * len)
|
||||||
|
copyMem(data, arr, sizeof(T) * len)
|
||||||
|
return (cast[ptr UncheckedArray[T]](data), len)
|
||||||
|
|||||||
@ -45,6 +45,13 @@ int UnwrapReceivedMessage(void* ctx,
|
|||||||
SdsCallBack callback,
|
SdsCallBack callback,
|
||||||
void* userData);
|
void* userData);
|
||||||
|
|
||||||
|
int MarkDependenciesMet(void* ctx,
|
||||||
|
char** messageIDs,
|
||||||
|
size_t count,
|
||||||
|
SdsCallBack callback,
|
||||||
|
void* userData);
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
#ifdef __cplusplus
|
#ifdef __cplusplus
|
||||||
}
|
}
|
||||||
|
|||||||
@ -14,7 +14,7 @@ import
|
|||||||
./ffi_types,
|
./ffi_types,
|
||||||
./sds_thread/inter_thread_communication/sds_thread_request,
|
./sds_thread/inter_thread_communication/sds_thread_request,
|
||||||
./sds_thread/inter_thread_communication/requests/
|
./sds_thread/inter_thread_communication/requests/
|
||||||
[sds_lifecycle_request, sds_message_request],
|
[sds_lifecycle_request, sds_message_request, sds_dependencies_request],
|
||||||
../src/[reliability, reliability_utils, message]
|
../src/[reliability, reliability_utils, message]
|
||||||
|
|
||||||
################################################################################
|
################################################################################
|
||||||
@ -242,5 +242,35 @@ proc UnwrapReceivedMessage(
|
|||||||
userData,
|
userData,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
proc MarkDependenciesMet(
|
||||||
|
ctx: ptr SdsContext,
|
||||||
|
messageIds: pointer,
|
||||||
|
count: csize_t,
|
||||||
|
callback: SdsCallBack,
|
||||||
|
userData: pointer,
|
||||||
|
): cint {.dynlib, exportc.} =
|
||||||
|
initializeLibrary()
|
||||||
|
checkLibsdsParams(ctx, callback, userData)
|
||||||
|
|
||||||
|
if messageIds == nil and count > 0:
|
||||||
|
let msg = "libsds error: " & "MessageIDs pointer is NULL but count > 0"
|
||||||
|
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](len(msg)), userData)
|
||||||
|
return RET_ERR
|
||||||
|
|
||||||
|
var messageIds = allocSharedSeqFromCArray(cast[ptr cstring](messageIds), count.int)
|
||||||
|
|
||||||
|
defer:
|
||||||
|
deallocSharedSeq(messageIds)
|
||||||
|
|
||||||
|
handleRequest(
|
||||||
|
ctx,
|
||||||
|
RequestType.DEPENDENCIES,
|
||||||
|
SdsDependenciesRequest.createShared(
|
||||||
|
SdsDependenciesMsgType.MARK_DEPENDENCIES_MET, messageIds, count
|
||||||
|
),
|
||||||
|
callback,
|
||||||
|
userData,
|
||||||
|
)
|
||||||
|
|
||||||
### End of exported procs
|
### End of exported procs
|
||||||
################################################################################
|
################################################################################
|
||||||
|
|||||||
@ -0,0 +1,49 @@
|
|||||||
|
import std/[options, json, strutils, net, sequtils]
|
||||||
|
import chronos, chronicles, results, confutils, confutils/std/net
|
||||||
|
|
||||||
|
import ../../../alloc
|
||||||
|
import ../../../../src/[reliability_utils, reliability, message]
|
||||||
|
|
||||||
|
type SdsDependenciesMsgType* = enum
|
||||||
|
MARK_DEPENDENCIES_MET
|
||||||
|
|
||||||
|
type SdsDependenciesRequest* = object
|
||||||
|
operation: SdsDependenciesMsgType
|
||||||
|
messageIds: SharedSeq[cstring]
|
||||||
|
count: csize_t
|
||||||
|
|
||||||
|
proc createShared*(
|
||||||
|
T: type SdsDependenciesRequest,
|
||||||
|
op: SdsDependenciesMsgType,
|
||||||
|
messageIds: SharedSeq[cstring],
|
||||||
|
count: csize_t = 0,
|
||||||
|
): ptr type T =
|
||||||
|
var ret = createShared(T)
|
||||||
|
ret[].operation = op
|
||||||
|
ret[].messageIds = messageIds # check if alloc is needed
|
||||||
|
ret[].count = count
|
||||||
|
return ret
|
||||||
|
|
||||||
|
proc destroyShared(self: ptr SdsDependenciesRequest) =
|
||||||
|
#deallocShared(self[].message)
|
||||||
|
#deallocShared(self[].messageId)
|
||||||
|
deallocShared(self)
|
||||||
|
|
||||||
|
proc process*(
|
||||||
|
self: ptr SdsDependenciesRequest, rm: ptr ReliabilityManager
|
||||||
|
): Future[Result[string, string]] {.async.} =
|
||||||
|
defer:
|
||||||
|
destroyShared(self)
|
||||||
|
|
||||||
|
case self.operation
|
||||||
|
of MARK_DEPENDENCIES_MET:
|
||||||
|
let messageIdsC = self.messageIds.toSeq()
|
||||||
|
let messageIds = messageIdsC.mapIt($it)
|
||||||
|
|
||||||
|
markDependenciesMet(rm[], messageIds).isOkOr:
|
||||||
|
error "MARK_DEPENDENCIES_MET failed", error = error
|
||||||
|
return err("error processing MARK_DEPENDENCIES_MET request: " & $error)
|
||||||
|
|
||||||
|
return ok("")
|
||||||
|
|
||||||
|
return ok("")
|
||||||
@ -6,12 +6,13 @@ import std/json, results
|
|||||||
import chronos, chronos/threadsync
|
import chronos, chronos/threadsync
|
||||||
import
|
import
|
||||||
../../ffi_types,
|
../../ffi_types,
|
||||||
./requests/[sds_lifecycle_request, sds_message_request],
|
./requests/[sds_lifecycle_request, sds_message_request, sds_dependencies_request],
|
||||||
../../../src/[reliability_utils]
|
../../../src/[reliability_utils]
|
||||||
|
|
||||||
type RequestType* {.pure.} = enum
|
type RequestType* {.pure.} = enum
|
||||||
LIFECYCLE
|
LIFECYCLE
|
||||||
MESSAGE
|
MESSAGE
|
||||||
|
DEPENDENCIES
|
||||||
|
|
||||||
type SdsThreadRequest* = object
|
type SdsThreadRequest* = object
|
||||||
reqType: RequestType
|
reqType: RequestType
|
||||||
@ -68,6 +69,8 @@ proc process*(
|
|||||||
cast[ptr SdsLifecycleRequest](request[].reqContent).process(rm)
|
cast[ptr SdsLifecycleRequest](request[].reqContent).process(rm)
|
||||||
of MESSAGE:
|
of MESSAGE:
|
||||||
cast[ptr SdsMessageRequest](request[].reqContent).process(rm)
|
cast[ptr SdsMessageRequest](request[].reqContent).process(rm)
|
||||||
|
of DEPENDENCIES:
|
||||||
|
cast[ptr SdsDependenciesRequest](request[].reqContent).process(rm)
|
||||||
|
|
||||||
handleRes(await retFut, request)
|
handleRes(await retFut, request)
|
||||||
|
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user