logos-messaging/nim-sds
1
0
Fork 0
mirror of https://github.com/logos-messaging/nim-sds.git synced 2026-01-07 16:43:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: libsds - support for multiple channels per RM (#15)

Browse Source
This commit is contained in:
Akhil 2025-07-17 14:13:21 +05:30 committed by GitHub
parent 8312df7b53
commit 4e10d77218
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 68 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
library/events/json_message_ready_event.nim
View File

@ -3,9 +3,10 @@ import ./json_base_event, ../../src/[message]
type JsonMessageReadyEvent* = ref object of JsonEvent type JsonMessageReadyEvent* = ref object of JsonEvent
messageId*: SdsMessageID messageId*: SdsMessageID
channelId*: SdsChannelID
proc new*(T: type JsonMessageReadyEvent, messageId: SdsMessageID): T = proc new*(T: type JsonMessageReadyEvent, messageId: SdsMessageID, channelId: SdsChannelID): T =
return JsonMessageReadyEvent(eventType: "message_ready", messageId: messageId) return JsonMessageReadyEvent(eventType: "message_ready", messageId: messageId, channelId: channelId)
method `$`*(jsonMessageReady: JsonMessageReadyEvent): string = method `$`*(jsonMessageReady: JsonMessageReadyEvent): string =
$(%*jsonMessageReady) $(%*jsonMessageReady)

5
library/events/json_message_sent_event.nim
View File

@ -3,9 +3,10 @@ import ./json_base_event, ../../src/[message]
type JsonMessageSentEvent* = ref object of JsonEvent type JsonMessageSentEvent* = ref object of JsonEvent
messageId*: SdsMessageID messageId*: SdsMessageID
channelId*: SdsChannelID
proc new*(T: type JsonMessageSentEvent, messageId: SdsMessageID): T = proc new*(T: type JsonMessageSentEvent, messageId: SdsMessageID, channelId: SdsChannelID): T =
return JsonMessageSentEvent(eventType: "message_sent", messageId: messageId) return JsonMessageSentEvent(eventType: "message_sent", messageId: messageId, channelId: channelId)
method `$`*(jsonMessageSent: JsonMessageSentEvent): string = method `$`*(jsonMessageSent: JsonMessageSentEvent): string =
$(%*jsonMessageSent) $(%*jsonMessageSent)

4
library/events/json_missing_dependencies_event.nim
View File

@ -4,14 +4,16 @@ import ./json_base_event, ../../src/[message]
type JsonMissingDependenciesEvent* = ref object of JsonEvent type JsonMissingDependenciesEvent* = ref object of JsonEvent
messageId*: SdsMessageID messageId*: SdsMessageID
missingDeps: seq[SdsMessageID] missingDeps: seq[SdsMessageID]
channelId*: SdsChannelID
proc new*( proc new*(
T: type JsonMissingDependenciesEvent, T: type JsonMissingDependenciesEvent,
messageId: SdsMessageID, messageId: SdsMessageID,
missingDeps: seq[SdsMessageID], missingDeps: seq[SdsMessageID],
channelId: SdsChannelID,
): T = ): T =
return JsonMissingDependenciesEvent( return JsonMissingDependenciesEvent(
eventType: "missing_dependencies", messageId: messageId, missingDeps: missingDeps eventType: "missing_dependencies", messageId: messageId, missingDeps: missingDeps, channelId: channelId
) )
method `$`*(jsonMissingDependencies: JsonMissingDependenciesEvent): string = method `$`*(jsonMissingDependencies: JsonMissingDependenciesEvent): string =

4
library/libsds.h
View File

@ -24,7 +24,7 @@ typedef void (*SdsCallBack) (int callerRet, const char* msg, size_t len, void* u
// --- Core API Functions --- // --- Core API Functions ---
void* SdsNewReliabilityManager(const char* channelId, SdsCallBack callback, void* userData); void* SdsNewReliabilityManager(SdsCallBack callback, void* userData);
void SdsSetEventCallback(void* ctx, SdsCallBack callback, void* userData); void SdsSetEventCallback(void* ctx, SdsCallBack callback, void* userData);
@ -36,6 +36,7 @@ int SdsWrapOutgoingMessage(void* ctx,
void* message, void* message,
size_t messageLen, size_t messageLen,
const char* messageId, const char* messageId,
const char* channelId,
SdsCallBack callback, SdsCallBack callback,
void* userData); void* userData);
@ -48,6 +49,7 @@ int SdsUnwrapReceivedMessage(void* ctx,
int SdsMarkDependenciesMet(void* ctx, int SdsMarkDependenciesMet(void* ctx,
char** messageIDs, char** messageIDs,
size_t count, size_t count,
const char* channelId,
SdsCallBack callback, SdsCallBack callback,
void* userData); void* userData);

46
library/libsds.nim
View File

@ -5,7 +5,7 @@
when defined(linux): when defined(linux):
{.passl: "-Wl,-soname,libsds.so".} {.passl: "-Wl,-soname,libsds.so".}
import std/[locks, typetraits, tables, atomics], chronos, chronicles import std/[typetraits, tables, atomics], chronos, chronicles
import import
./sds_thread/sds_thread, ./sds_thread/sds_thread,
./alloc, ./alloc,
@ -13,7 +13,7 @@ import
./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_dependencies_request], [sds_lifecycle_request, sds_message_request, sds_dependencies_request],
../src/[reliability, reliability_utils, message], ../src/[reliability_utils, message],
./events/[ ./events/[
json_message_ready_event, json_message_sent_event, json_missing_dependencies_event, json_message_ready_event, json_message_sent_event, json_missing_dependencies_event,
json_periodic_sync_event, json_periodic_sync_event,
@ -72,19 +72,19 @@ proc handleRequest(
return RET_OK return RET_OK
proc onMessageReady(ctx: ptr SdsContext): MessageReadyCallback = proc onMessageReady(ctx: ptr SdsContext): MessageReadyCallback =
return proc(messageId: SdsMessageID) {.gcsafe.} = return proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} =
callEventCallback(ctx, "onMessageReady"): callEventCallback(ctx, "onMessageReady"):
$JsonMessageReadyEvent.new(messageId) $JsonMessageReadyEvent.new(messageId, channelId)
proc onMessageSent(ctx: ptr SdsContext): MessageSentCallback = proc onMessageSent(ctx: ptr SdsContext): MessageSentCallback =
return proc(messageId: SdsMessageID) {.gcsafe.} = return proc(messageId: SdsMessageID, channelId: SdsChannelID) {.gcsafe.} =
callEventCallback(ctx, "onMessageSent"): callEventCallback(ctx, "onMessageSent"):
$JsonMessageSentEvent.new(messageId) $JsonMessageSentEvent.new(messageId, channelId)
proc onMissingDependencies(ctx: ptr SdsContext): MissingDependenciesCallback = proc onMissingDependencies(ctx: ptr SdsContext): MissingDependenciesCallback =
return proc(messageId: SdsMessageID, missingDeps: seq[SdsMessageID]) {.gcsafe.} = return proc(messageId: SdsMessageID, missingDeps: seq[SdsMessageID], channelId: SdsChannelID) {.gcsafe.} =
callEventCallback(ctx, "onMissingDependencies"): callEventCallback(ctx, "onMissingDependencies"):
$JsonMissingDependenciesEvent.new(messageId, missingDeps) $JsonMissingDependenciesEvent.new(messageId, missingDeps, channelId)
proc onPeriodicSync(ctx: ptr SdsContext): PeriodicSyncCallback = proc onPeriodicSync(ctx: ptr SdsContext): PeriodicSyncCallback =
return proc() {.gcsafe.} = return proc() {.gcsafe.} =
@ -131,7 +131,7 @@ proc initializeLibrary() {.exported.} =
### Exported procs ### Exported procs
proc SdsNewReliabilityManager( proc SdsNewReliabilityManager(
channelId: cstring, callback: SdsCallBack, userData: pointer callback: SdsCallBack, userData: pointer
): pointer {.dynlib, exportc, cdecl.} = ): pointer {.dynlib, exportc, cdecl.} =
initializeLibrary() initializeLibrary()
@ -159,7 +159,7 @@ proc SdsNewReliabilityManager(
ctx, ctx,
RequestType.LIFECYCLE, RequestType.LIFECYCLE,
SdsLifecycleRequest.createShared( SdsLifecycleRequest.createShared(
SdsLifecycleMsgType.CREATE_RELIABILITY_MANAGER, channelId, appCallbacks SdsLifecycleMsgType.CREATE_RELIABILITY_MANAGER, nil, appCallbacks
), ),
callback, callback,
userData, userData,
@ -211,6 +211,7 @@ proc SdsWrapOutgoingMessage(
message: pointer, message: pointer,
messageLen: csize_t, messageLen: csize_t,
messageId: cstring, messageId: cstring,
channelId: cstring,
callback: SdsCallBack, callback: SdsCallBack,
userData: pointer, userData: pointer,
): cint {.dynlib, exportc.} = ): cint {.dynlib, exportc.} =
@ -227,11 +228,21 @@ proc SdsWrapOutgoingMessage(
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](len(msg)), userData) callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](len(msg)), userData)
return RET_ERR return RET_ERR
if channelId == nil:
let msg = "libsds error: " & "channel ID pointer is NULL"
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](len(msg)), userData)
return RET_ERR
if channelId != nil and $channelId == "":
let msg = "libsds error: " & "channel ID is empty string"
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](len(msg)), userData)
return RET_ERR
handleRequest( handleRequest(
ctx, ctx,
RequestType.MESSAGE, RequestType.MESSAGE,
SdsMessageRequest.createShared( SdsMessageRequest.createShared(
SdsMessageMsgType.WRAP_MESSAGE, message, messageLen, messageId SdsMessageMsgType.WRAP_MESSAGE, message, messageLen, messageId, channelId
), ),
callback, callback,
userData, userData,
@ -266,6 +277,7 @@ proc SdsMarkDependenciesMet(
ctx: ptr SdsContext, ctx: ptr SdsContext,
messageIds: pointer, messageIds: pointer,
count: csize_t, count: csize_t,
channelId: cstring,
callback: SdsCallBack, callback: SdsCallBack,
userData: pointer, userData: pointer,
): cint {.dynlib, exportc.} = ): cint {.dynlib, exportc.} =
@ -277,11 +289,21 @@ proc SdsMarkDependenciesMet(
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](len(msg)), userData) callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](len(msg)), userData)
return RET_ERR return RET_ERR
if channelId == nil:
let msg = "libsds error: " & "channel ID pointer is NULL"
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](len(msg)), userData)
return RET_ERR
if channelId != nil and $channelId == "":
let msg = "libsds error: " & "channel ID is empty string"
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](len(msg)), userData)
return RET_ERR
handleRequest( handleRequest(
ctx, ctx,
RequestType.DEPENDENCIES, RequestType.DEPENDENCIES,
SdsDependenciesRequest.createShared( SdsDependenciesRequest.createShared(
SdsDependenciesMsgType.MARK_DEPENDENCIES_MET, messageIds, count SdsDependenciesMsgType.MARK_DEPENDENCIES_MET, messageIds, count, channelId
), ),
callback, callback,
userData, userData,

12
library/sds_thread/inter_thread_communication/requests/sds_dependencies_request.nim
View File

@ -1,8 +1,8 @@
import std/[options, json, strutils, net, sequtils] import std/[json, strutils, net, sequtils]
import chronos, chronicles, results, confutils, confutils/std/net import chronos, chronicles, results
import ../../../alloc import ../../../alloc
import ../../../../src/[reliability_utils, reliability, message] import ../../../../src/[reliability_utils, reliability]
type SdsDependenciesMsgType* = enum type SdsDependenciesMsgType* = enum
MARK_DEPENDENCIES_MET MARK_DEPENDENCIES_MET
@ -11,21 +11,25 @@ type SdsDependenciesRequest* = object
operation: SdsDependenciesMsgType operation: SdsDependenciesMsgType
messageIds: SharedSeq[cstring] messageIds: SharedSeq[cstring]
count: csize_t count: csize_t
channelId: cstring
proc createShared*( proc createShared*(
T: type SdsDependenciesRequest, T: type SdsDependenciesRequest,
op: SdsDependenciesMsgType, op: SdsDependenciesMsgType,
messageIds: pointer, messageIds: pointer,
count: csize_t = 0, count: csize_t = 0,
channelId: cstring = "",
): ptr type T = ): ptr type T =
var ret = createShared(T) var ret = createShared(T)
ret[].operation = op ret[].operation = op
ret[].count = count ret[].count = count
ret[].channelId = channelId.alloc()
ret[].messageIds = allocSharedSeqFromCArray(cast[ptr cstring](messageIds), count.int) ret[].messageIds = allocSharedSeqFromCArray(cast[ptr cstring](messageIds), count.int)
return ret return ret
proc destroyShared(self: ptr SdsDependenciesRequest) = proc destroyShared(self: ptr SdsDependenciesRequest) =
deallocSharedSeq(self[].messageIds) deallocSharedSeq(self[].messageIds)
deallocShared(self[].channelId)
deallocShared(self) deallocShared(self)
proc process*( proc process*(
@ -39,7 +43,7 @@ proc process*(
let messageIdsC = self.messageIds.toSeq() let messageIdsC = self.messageIds.toSeq()
let messageIds = messageIdsC.mapIt($it) let messageIds = messageIdsC.mapIt($it)
markDependenciesMet(rm[], messageIds).isOkOr: markDependenciesMet(rm[], messageIds, $self.channelId).isOkOr:
error "MARK_DEPENDENCIES_MET failed", error = error error "MARK_DEPENDENCIES_MET failed", error = error
return err("error processing MARK_DEPENDENCIES_MET request: " & $error) return err("error processing MARK_DEPENDENCIES_MET request: " & $error)

17
library/sds_thread/inter_thread_communication/requests/sds_lifecycle_request.nim
View File

@ -1,8 +1,8 @@
import std/[options, json, strutils, net] import std/json
import chronos, chronicles, results, confutils, confutils/std/net import chronos, chronicles, results
import ../../../alloc import ../../../alloc
import ../../../../src/[reliability_utils, reliability, message] import ../../../../src/[reliability_utils, reliability]
type SdsLifecycleMsgType* = enum type SdsLifecycleMsgType* = enum
CREATE_RELIABILITY_MANAGER CREATE_RELIABILITY_MANAGER
@ -31,14 +31,9 @@ proc destroyShared(self: ptr SdsLifecycleRequest) =
deallocShared(self) deallocShared(self)
proc createReliabilityManager( proc createReliabilityManager(
channelIdCStr: cstring, appCallbacks: AppCallbacks = nil appCallbacks: AppCallbacks = nil
): Future[Result[ReliabilityManager, string]] {.async.} = ): Future[Result[ReliabilityManager, string]] {.async.} =
let channelId = $channelIdCStr let rm = newReliabilityManager().valueOr:
if channelId.len == 0:
error "Failed creating ReliabilityManager: Channel ID cannot be empty"
return err("Failed creating ReliabilityManager: Channel ID cannot be empty")
let rm = newReliabilityManager(some(channelId)).valueOr:
error "Failed creating reliability manager", error = error error "Failed creating reliability manager", error = error
return err("Failed creating reliability manager: " & $error) return err("Failed creating reliability manager: " & $error)
@ -57,7 +52,7 @@ proc process*(
case self.operation case self.operation
of CREATE_RELIABILITY_MANAGER: of CREATE_RELIABILITY_MANAGER:
rm[] = (await createReliabilityManager(self.channelId, self.appCallbacks)).valueOr: rm[] = (await createReliabilityManager(self.appCallbacks)).valueOr:
error "CREATE_RELIABILITY_MANAGER failed", error = error error "CREATE_RELIABILITY_MANAGER failed", error = error
return err("error processing CREATE_RELIABILITY_MANAGER request: " & $error) return err("error processing CREATE_RELIABILITY_MANAGER request: " & $error)
of RESET_RELIABILITY_MANAGER: of RESET_RELIABILITY_MANAGER:

12
library/sds_thread/inter_thread_communication/requests/sds_message_request.nim
View File

@ -1,5 +1,5 @@
import std/[options, json, strutils, net, sequtils] import std/[json, strutils, net, sequtils]
import chronos, chronicles, results, confutils, confutils/std/net import chronos, chronicles, results
import ../../../alloc import ../../../alloc
import ../../../../src/[reliability_utils, reliability, message] import ../../../../src/[reliability_utils, reliability, message]
@ -13,6 +13,7 @@ type SdsMessageRequest* = object
message: SharedSeq[byte] message: SharedSeq[byte]
messageLen: csize_t messageLen: csize_t
messageId: cstring messageId: cstring
channelId: cstring
type SdsUnwrapResponse* = object type SdsUnwrapResponse* = object
message*: seq[byte] message*: seq[byte]
@ -24,11 +25,13 @@ proc createShared*(
message: pointer, message: pointer,
messageLen: csize_t = 0, messageLen: csize_t = 0,
messageId: cstring = "", messageId: cstring = "",
channelId: cstring = "",
): ptr type T = ): ptr type T =
var ret = createShared(T) var ret = createShared(T)
ret[].operation = op ret[].operation = op
ret[].messageLen = messageLen ret[].messageLen = messageLen
ret[].messageId = messageId.alloc() ret[].messageId = messageId.alloc()
ret[].channelId = channelId.alloc()
ret[].message = allocSharedSeqFromCArray(cast[ptr byte](message), messageLen.int) ret[].message = allocSharedSeqFromCArray(cast[ptr byte](message), messageLen.int)
return ret return ret
@ -36,6 +39,7 @@ proc createShared*(
proc destroyShared(self: ptr SdsMessageRequest) = proc destroyShared(self: ptr SdsMessageRequest) =
deallocSharedSeq(self[].message) deallocSharedSeq(self[].message)
deallocShared(self[].messageId) deallocShared(self[].messageId)
deallocShared(self[].channelId)
deallocShared(self) deallocShared(self)
proc process*( proc process*(
@ -48,7 +52,7 @@ proc process*(
of WRAP_MESSAGE: of WRAP_MESSAGE:
let messageBytes = self.message.toSeq() let messageBytes = self.message.toSeq()
let wrappedMessage = wrapOutgoingMessage(rm[], messageBytes, $self.messageId).valueOr: let wrappedMessage = wrapOutgoingMessage(rm[], messageBytes, $self.messageId, $self.channelId).valueOr:
error "WRAP_MESSAGE failed", error = error error "WRAP_MESSAGE failed", error = error
return err("error processing WRAP_MESSAGE request: " & $error) return err("error processing WRAP_MESSAGE request: " & $error)
@ -57,7 +61,7 @@ proc process*(
of UNWRAP_MESSAGE: of UNWRAP_MESSAGE:
let messageBytes = self.message.toSeq() let messageBytes = self.message.toSeq()
let (unwrappedMessage, missingDeps) = unwrapReceivedMessage(rm[], messageBytes).valueOr: let (unwrappedMessage, missingDeps, _) = unwrapReceivedMessage(rm[], messageBytes).valueOr:
error "UNWRAP_MESSAGE failed", error = error error "UNWRAP_MESSAGE failed", error = error
return err("error processing UNWRAP_MESSAGE request: " & $error) return err("error processing UNWRAP_MESSAGE request: " & $error)