mirror of
https://github.com/logos-messaging/sds-go-bindings.git
synced 2026-05-05 01:53:30 +00:00
848 lines
25 KiB
Go
848 lines
25 KiB
Go
package sds
|
|
|
|
import (
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
// Test basic creation, cleanup, and reset
|
|
func TestLifecycle(t *testing.T) {
|
|
rm, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
require.NotNil(t, rm, "Expected ReliabilityManager to be not nil")
|
|
|
|
defer rm.Cleanup() // Ensure cleanup even on test failure
|
|
|
|
err = rm.Reset()
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
// Test wrapping and unwrapping a simple message
|
|
func TestWrapUnwrap(t *testing.T) {
|
|
rm, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer rm.Cleanup()
|
|
|
|
channelID := "test-wrap-unwrap"
|
|
originalPayload := []byte("hello reliability")
|
|
messageID := MessageID("msg-wrap-1")
|
|
|
|
wrappedMsg, err := rm.WrapOutgoingMessage(originalPayload, messageID, channelID)
|
|
require.NoError(t, err)
|
|
|
|
require.Greater(t, len(wrappedMsg), 0, "Expected non-empty wrapped message")
|
|
|
|
// Simulate receiving the wrapped message
|
|
unwrappedMessage, err := rm.UnwrapReceivedMessage(wrappedMsg)
|
|
require.NoError(t, err)
|
|
|
|
require.Equal(t, string(*unwrappedMessage.Message), string(originalPayload), "Expected unwrapped and original payloads to be equal")
|
|
require.Equal(t, len(*unwrappedMessage.MissingDeps), 0, "Expexted to be no missing dependencies")
|
|
}
|
|
|
|
// Test dependency handling
|
|
func TestDependencies(t *testing.T) {
|
|
rm, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer rm.Cleanup()
|
|
|
|
channelID := "test-deps"
|
|
|
|
// 1. Send message 1 (will become a dependency)
|
|
payload1 := []byte("message one")
|
|
msgID1 := MessageID("msg-dep-1")
|
|
wrappedMsg1, err := rm.WrapOutgoingMessage(payload1, msgID1, channelID)
|
|
require.NoError(t, err)
|
|
|
|
// Simulate receiving msg1 to add it to history (implicitly acknowledges it)
|
|
_, err = rm.UnwrapReceivedMessage(wrappedMsg1)
|
|
require.NoError(t, err)
|
|
|
|
// 2. Send message 2 (depends on message 1 implicitly via causal history)
|
|
payload2 := []byte("message two")
|
|
msgID2 := MessageID("msg-dep-2")
|
|
wrappedMsg2, err := rm.WrapOutgoingMessage(payload2, msgID2, channelID)
|
|
require.NoError(t, err)
|
|
|
|
// 3. Create a new manager to simulate a different peer receiving msg2 without msg1
|
|
rm2, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer rm2.Cleanup()
|
|
|
|
// 4. Unwrap message 2 on the second manager - should report msg1 as missing
|
|
unwrappedMessage2, err := rm2.UnwrapReceivedMessage(wrappedMsg2)
|
|
require.NoError(t, err)
|
|
|
|
require.Greater(t, len(*unwrappedMessage2.MissingDeps), 0, "Expected missing dependencies, got none")
|
|
|
|
foundDep1 := false
|
|
for _, dep := range *unwrappedMessage2.MissingDeps {
|
|
if dep == msgID1 {
|
|
foundDep1 = true
|
|
break
|
|
}
|
|
}
|
|
require.True(t, foundDep1, "Expected missing dependency %q, got %v", msgID1, *unwrappedMessage2.MissingDeps)
|
|
|
|
// 5. Mark the dependency as met
|
|
err = rm2.MarkDependenciesMet([]MessageID{msgID1}, channelID)
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
// Test OnMessageReady callback
|
|
func TestCallback_OnMessageReady(t *testing.T) {
|
|
// Create sender and receiver RMs
|
|
senderRm, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer senderRm.Cleanup()
|
|
|
|
receiverRm, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer receiverRm.Cleanup()
|
|
|
|
channelID := "test-cb-ready"
|
|
|
|
// Use a channel for signaling
|
|
readyChan := make(chan MessageID, 1)
|
|
|
|
callbacks := EventCallbacks{
|
|
OnMessageReady: func(messageId MessageID, chId string) {
|
|
require.Equal(t, channelID, chId)
|
|
// Non-blocking send to channel
|
|
select {
|
|
case readyChan <- messageId:
|
|
default:
|
|
// Avoid blocking if channel is full or test already timed out
|
|
}
|
|
},
|
|
}
|
|
|
|
// Register callback only on the receiver
|
|
receiverRm.RegisterCallbacks(callbacks)
|
|
|
|
// Scenario: Wrap message on sender, unwrap on receiver
|
|
payload := []byte("ready test")
|
|
msgID := MessageID("cb-ready-1")
|
|
|
|
// Wrap on sender
|
|
wrappedMsg, err := senderRm.WrapOutgoingMessage(payload, msgID, channelID)
|
|
require.NoError(t, err)
|
|
|
|
// Unwrap on receiver
|
|
_, err = receiverRm.UnwrapReceivedMessage(wrappedMsg)
|
|
require.NoError(t, err)
|
|
|
|
// Verification - Wait on channel with timeout
|
|
select {
|
|
case receivedMsgID := <-readyChan:
|
|
// Mark as called implicitly since we received on channel
|
|
if receivedMsgID != msgID {
|
|
t.Errorf("OnMessageReady called with wrong ID: got %q, want %q", receivedMsgID, msgID)
|
|
}
|
|
case <-time.After(2 * time.Second):
|
|
// If timeout occurs, the channel receive failed.
|
|
t.Errorf("Timed out waiting for OnMessageReady callback on readyChan")
|
|
}
|
|
}
|
|
|
|
// Test OnMessageSent callback (via causal history ACK)
|
|
func TestCallback_OnMessageSent(t *testing.T) {
|
|
// Create two RMs
|
|
rm1, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer rm1.Cleanup()
|
|
|
|
rm2, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer rm2.Cleanup()
|
|
|
|
channelID := "test-cb-sent"
|
|
|
|
var wg sync.WaitGroup
|
|
sentCalled := false
|
|
var sentMsgID MessageID
|
|
var cbMutex sync.Mutex
|
|
|
|
callbacks := EventCallbacks{
|
|
OnMessageSent: func(messageId MessageID, chId string) {
|
|
require.Equal(t, channelID, chId)
|
|
cbMutex.Lock()
|
|
sentCalled = true
|
|
sentMsgID = messageId
|
|
cbMutex.Unlock()
|
|
wg.Done()
|
|
},
|
|
}
|
|
|
|
// Register callback on rm1 (the original sender)
|
|
rm1.RegisterCallbacks(callbacks)
|
|
|
|
// Scenario: rm1 sends msg1, rm2 receives msg1,
|
|
// rm2 sends msg2 (acking msg1), rm1 receives msg2.
|
|
|
|
// 1. rm1 sends msg1
|
|
payload1 := []byte("sent test 1")
|
|
msgID1 := MessageID("cb-sent-1")
|
|
wrappedMsg1, err := rm1.WrapOutgoingMessage(payload1, msgID1, channelID)
|
|
require.NoError(t, err)
|
|
// Note: msg1 is now in rm1's outgoing buffer
|
|
|
|
// 2. rm2 receives msg1 (to update its state)
|
|
_, err = rm2.UnwrapReceivedMessage(wrappedMsg1)
|
|
require.NoError(t, err)
|
|
|
|
// 3. rm2 sends msg2 (will include msg1 in causal history)
|
|
payload2 := []byte("sent test 2")
|
|
msgID2 := MessageID("cb-sent-2")
|
|
wrappedMsg2, err := rm2.WrapOutgoingMessage(payload2, msgID2, channelID)
|
|
require.NoError(t, err)
|
|
|
|
// 4. rm1 receives msg2 (should trigger ACK for msg1)
|
|
wg.Add(1) // Expect OnMessageSent for msg1 on rm1
|
|
_, err = rm1.UnwrapReceivedMessage(wrappedMsg2)
|
|
require.NoError(t, err)
|
|
|
|
// Verification
|
|
waitTimeout(&wg, 2*time.Second, t)
|
|
|
|
cbMutex.Lock()
|
|
defer cbMutex.Unlock()
|
|
if !sentCalled {
|
|
t.Errorf("OnMessageSent was not called")
|
|
}
|
|
// We primarily care that msg1 was ACKed.
|
|
if sentMsgID != msgID1 {
|
|
t.Errorf("OnMessageSent called with wrong ID: got %q, want %q", sentMsgID, msgID1)
|
|
}
|
|
}
|
|
|
|
// Test OnMissingDependencies callback
|
|
func TestCallback_OnMissingDependencies(t *testing.T) {
|
|
// Use separate sender/receiver RMs explicitly
|
|
senderRm, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer senderRm.Cleanup()
|
|
|
|
receiverRm, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer receiverRm.Cleanup()
|
|
|
|
channelID := "test-cb-missing"
|
|
|
|
var wg sync.WaitGroup
|
|
missingCalled := false
|
|
var missingMsgID MessageID
|
|
var missingDepsList []MessageID
|
|
var cbMutex sync.Mutex
|
|
|
|
callbacks := EventCallbacks{
|
|
OnMissingDependencies: func(messageId MessageID, missingDeps []MessageID, chId string) {
|
|
require.Equal(t, channelID, chId)
|
|
cbMutex.Lock()
|
|
missingCalled = true
|
|
missingMsgID = messageId
|
|
missingDepsList = missingDeps // Copy slice
|
|
cbMutex.Unlock()
|
|
wg.Done()
|
|
},
|
|
}
|
|
|
|
// Register callback only on the receiver rm
|
|
receiverRm.RegisterCallbacks(callbacks)
|
|
|
|
// Scenario: Sender sends msg1, then sender sends msg2 (depends on msg1),
|
|
// then receiver receives msg2 (which hasn't seen msg1).
|
|
|
|
// 1. Sender sends msg1
|
|
payload1 := []byte("missing test 1")
|
|
msgID1 := MessageID("cb-miss-1")
|
|
_, err = senderRm.WrapOutgoingMessage(payload1, msgID1, channelID)
|
|
require.NoError(t, err)
|
|
|
|
// 2. Sender sends msg2 (depends on msg1)
|
|
payload2 := []byte("missing test 2")
|
|
msgID2 := MessageID("cb-miss-2")
|
|
wrappedMsg2, err := senderRm.WrapOutgoingMessage(payload2, msgID2, channelID)
|
|
require.NoError(t, err)
|
|
|
|
// 3. Receiver receives msg2 (haven't seen msg1)
|
|
wg.Add(1) // Expect OnMissingDependencies
|
|
_, err = receiverRm.UnwrapReceivedMessage(wrappedMsg2)
|
|
require.NoError(t, err)
|
|
|
|
// Verification
|
|
waitTimeout(&wg, 2*time.Second, t)
|
|
|
|
cbMutex.Lock()
|
|
defer cbMutex.Unlock()
|
|
if !missingCalled {
|
|
t.Errorf("OnMissingDependencies was not called")
|
|
}
|
|
if missingMsgID != msgID2 {
|
|
t.Errorf("OnMissingDependencies called for wrong ID: got %q, want %q", missingMsgID, msgID2)
|
|
}
|
|
foundDep := false
|
|
for _, dep := range missingDepsList {
|
|
if dep == msgID1 {
|
|
foundDep = true
|
|
break
|
|
}
|
|
}
|
|
if !foundDep {
|
|
t.Errorf("OnMissingDependencies did not report %q as missing, got: %v", msgID1, missingDepsList)
|
|
}
|
|
}
|
|
|
|
// Test OnPeriodicSync callback
|
|
func TestCallback_OnPeriodicSync(t *testing.T) {
|
|
rm, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer rm.Cleanup()
|
|
|
|
syncCalled := false
|
|
var cbMutex sync.Mutex
|
|
// Use a channel to signal when the callback is hit
|
|
syncChan := make(chan bool, 1)
|
|
|
|
callbacks := EventCallbacks{
|
|
OnPeriodicSync: func() {
|
|
cbMutex.Lock()
|
|
if !syncCalled { // Only signal the first time
|
|
syncCalled = true
|
|
syncChan <- true
|
|
}
|
|
cbMutex.Unlock()
|
|
},
|
|
}
|
|
|
|
rm.RegisterCallbacks(callbacks)
|
|
|
|
// Start periodic tasks
|
|
err = rm.StartPeriodicTasks()
|
|
require.NoError(t, err)
|
|
|
|
// --- Verification ---
|
|
// Wait for the periodic sync callback with a timeout (needs to be longer than sync interval)
|
|
select {
|
|
case <-syncChan:
|
|
// Success
|
|
case <-time.After(10 * time.Second):
|
|
t.Errorf("Timed out waiting for OnPeriodicSync callback")
|
|
}
|
|
|
|
cbMutex.Lock()
|
|
defer cbMutex.Unlock()
|
|
if !syncCalled {
|
|
// This might happen if the timeout was too short
|
|
t.Logf("Warning: OnPeriodicSync might not have been called within the test timeout")
|
|
}
|
|
}
|
|
|
|
// Combined Test for multiple callbacks
|
|
func TestCallbacks_Combined(t *testing.T) {
|
|
// Create sender and receiver RMs
|
|
senderRm, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer senderRm.Cleanup()
|
|
|
|
receiverRm, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer receiverRm.Cleanup()
|
|
|
|
channelID := "test-cb-combined"
|
|
|
|
// Channels for synchronization
|
|
readyChan1 := make(chan bool, 1)
|
|
sentChan1 := make(chan bool, 1)
|
|
missingChan := make(chan []MessageID, 1)
|
|
|
|
// Use maps for verification
|
|
receivedReady := make(map[MessageID]bool)
|
|
receivedSent := make(map[MessageID]bool)
|
|
var cbMutex sync.Mutex
|
|
|
|
callbacksReceiver := EventCallbacks{
|
|
OnMessageReady: func(messageId MessageID, chId string) {
|
|
require.Equal(t, channelID, chId)
|
|
cbMutex.Lock()
|
|
receivedReady[messageId] = true
|
|
cbMutex.Unlock()
|
|
if messageId == "cb-comb-1" {
|
|
// Use non-blocking send
|
|
select {
|
|
case readyChan1 <- true:
|
|
default:
|
|
}
|
|
}
|
|
},
|
|
OnMessageSent: func(messageId MessageID, chId string) {
|
|
// This callback is registered on Receiver, but Sent events
|
|
// are typically relevant to the Sender. We don't expect this.
|
|
t.Errorf("Unexpected OnMessageSent call on Receiver for %s", messageId)
|
|
},
|
|
OnMissingDependencies: func(messageId MessageID, missingDeps []MessageID, chId string) {
|
|
// This callback is registered on Receiver, used for receiverRm2 below
|
|
},
|
|
}
|
|
|
|
callbacksSender := EventCallbacks{
|
|
OnMessageReady: func(messageId MessageID, chId string) {
|
|
// Not expected on sender in this test flow
|
|
},
|
|
OnMessageSent: func(messageId MessageID, chId string) {
|
|
require.Equal(t, channelID, chId)
|
|
cbMutex.Lock()
|
|
receivedSent[messageId] = true
|
|
cbMutex.Unlock()
|
|
if messageId == "cb-comb-1" {
|
|
select {
|
|
case sentChan1 <- true:
|
|
default:
|
|
}
|
|
}
|
|
},
|
|
OnMissingDependencies: func(messageId MessageID, missingDeps []MessageID, chId string) {
|
|
// Not expected on sender
|
|
},
|
|
}
|
|
|
|
// Register callbacks
|
|
receiverRm.RegisterCallbacks(callbacksReceiver)
|
|
senderRm.RegisterCallbacks(callbacksSender)
|
|
|
|
// --- Test Scenario ---
|
|
msgID1 := MessageID("cb-comb-1")
|
|
msgID2 := MessageID("cb-comb-2")
|
|
msgID3 := MessageID("cb-comb-3")
|
|
payload1 := []byte("combined test 1")
|
|
payload2 := []byte("combined test 2")
|
|
payload3 := []byte("combined test 3")
|
|
|
|
// 1. Sender sends msg1
|
|
wrappedMsg1, err := senderRm.WrapOutgoingMessage(payload1, msgID1, channelID)
|
|
require.NoError(t, err)
|
|
|
|
// 2. Receiver receives msg1
|
|
_, err = receiverRm.UnwrapReceivedMessage(wrappedMsg1)
|
|
require.NoError(t, err)
|
|
|
|
// 3. Receiver sends msg2 (depends on msg1 implicitly via state)
|
|
wrappedMsg2, err := receiverRm.WrapOutgoingMessage(payload2, msgID2, channelID)
|
|
require.NoError(t, err)
|
|
|
|
// 4. Sender receives msg2 from Receiver (acks msg1 for sender)
|
|
_, err = senderRm.UnwrapReceivedMessage(wrappedMsg2)
|
|
require.NoError(t, err)
|
|
|
|
// 5. Sender sends msg3 (depends on msg2)
|
|
wrappedMsg3, err := senderRm.WrapOutgoingMessage(payload3, msgID3, channelID)
|
|
require.NoError(t, err)
|
|
|
|
// 6. Create Receiver2, register missing deps callback
|
|
receiverRm2, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer receiverRm2.Cleanup()
|
|
|
|
callbacksReceiver2 := EventCallbacks{
|
|
OnMissingDependencies: func(messageId MessageID, missingDeps []MessageID, chId string) {
|
|
require.Equal(t, channelID, chId)
|
|
if messageId == msgID3 {
|
|
select {
|
|
case missingChan <- missingDeps:
|
|
default:
|
|
}
|
|
}
|
|
},
|
|
}
|
|
|
|
receiverRm2.RegisterCallbacks(callbacksReceiver2)
|
|
|
|
// 7. Receiver2 receives msg3 (should report missing msg1, msg2)
|
|
_, err = receiverRm2.UnwrapReceivedMessage(wrappedMsg3)
|
|
require.NoError(t, err)
|
|
|
|
// --- Verification ---
|
|
timeout := 5 * time.Second
|
|
expectedReady1 := false
|
|
expectedSent1 := false
|
|
var reportedMissingDeps []MessageID
|
|
missingDepsReceived := false
|
|
|
|
receivedCount := 0
|
|
expectedCount := 3 // ready1, sent1, missingDeps
|
|
timer := time.NewTimer(timeout)
|
|
defer timer.Stop()
|
|
|
|
for receivedCount < expectedCount {
|
|
select {
|
|
case <-readyChan1:
|
|
if !expectedReady1 { // Avoid double counting if signaled twice
|
|
expectedReady1 = true
|
|
receivedCount++
|
|
}
|
|
case <-sentChan1:
|
|
if !expectedSent1 {
|
|
expectedSent1 = true
|
|
receivedCount++
|
|
}
|
|
case deps := <-missingChan:
|
|
if !missingDepsReceived {
|
|
reportedMissingDeps = deps
|
|
missingDepsReceived = true
|
|
receivedCount++
|
|
}
|
|
case <-timer.C:
|
|
t.Fatalf("Timed out waiting for combined callbacks (received %d out of %d)", receivedCount, expectedCount)
|
|
}
|
|
}
|
|
|
|
// Check results
|
|
cbMutex.Lock()
|
|
defer cbMutex.Unlock()
|
|
|
|
if !expectedReady1 || !receivedReady[msgID1] {
|
|
t.Errorf("OnMessageReady not called/verified for %s", msgID1)
|
|
}
|
|
if !expectedSent1 || !receivedSent[msgID1] {
|
|
t.Errorf("OnMessageSent not called/verified for %s", msgID1)
|
|
}
|
|
if !missingDepsReceived {
|
|
t.Errorf("OnMissingDependencies not called/verified for %s", msgID3)
|
|
} else {
|
|
foundDep1 := false
|
|
foundDep2 := false
|
|
for _, dep := range reportedMissingDeps {
|
|
if dep == msgID1 {
|
|
foundDep1 = true
|
|
}
|
|
if dep == msgID2 {
|
|
foundDep2 = true
|
|
}
|
|
}
|
|
if !foundDep1 || !foundDep2 {
|
|
t.Errorf("OnMissingDependencies for %s reported wrong deps: got %v, want %s and %s", msgID3, reportedMissingDeps, msgID1, msgID2)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Helper function to wait for WaitGroup with a timeout
|
|
func waitTimeout(wg *sync.WaitGroup, timeout time.Duration, t *testing.T) {
|
|
c := make(chan struct{})
|
|
go func() {
|
|
defer close(c)
|
|
wg.Wait()
|
|
}()
|
|
select {
|
|
case <-c:
|
|
// Completed normally
|
|
case <-time.After(timeout):
|
|
t.Fatalf("Timed out waiting for callbacks")
|
|
}
|
|
}
|
|
|
|
// Test multi-channel functionality - one RM can handle messages from different channels
|
|
func TestMultiChannel_SingleRM(t *testing.T) {
|
|
rm, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer rm.Cleanup()
|
|
|
|
// Test with two different channels
|
|
channel1 := "test-channel-1"
|
|
channel2 := "test-channel-2"
|
|
|
|
// Create and wrap messages for different channels
|
|
msg1 := []byte("message for channel 1")
|
|
msgID1 := MessageID("msg1")
|
|
wrappedMsg1, err := rm.WrapOutgoingMessage(msg1, msgID1, channel1)
|
|
require.NoError(t, err)
|
|
|
|
msg2 := []byte("message for channel 2")
|
|
msgID2 := MessageID("msg2")
|
|
wrappedMsg2, err := rm.WrapOutgoingMessage(msg2, msgID2, channel2)
|
|
require.NoError(t, err)
|
|
|
|
// Unwrap messages - should extract channel ID and route correctly
|
|
unwrapped1, err := rm.UnwrapReceivedMessage(wrappedMsg1)
|
|
require.NoError(t, err)
|
|
require.Equal(t, msg1, *unwrapped1.Message)
|
|
|
|
// Verify channel ID is extracted correctly
|
|
require.NotNil(t, unwrapped1.ChannelId, "Expected ChannelId to be not nil")
|
|
require.Equal(t, channel1, *unwrapped1.ChannelId)
|
|
|
|
unwrapped2, err := rm.UnwrapReceivedMessage(wrappedMsg2)
|
|
require.NoError(t, err)
|
|
require.Equal(t, msg2, *unwrapped2.Message)
|
|
|
|
// Verify channel ID is extracted correctly
|
|
require.NotNil(t, unwrapped2.ChannelId, "Expected ChannelId to be not nil")
|
|
require.Equal(t, channel2, *unwrapped2.ChannelId)
|
|
|
|
// Test that the same RM can handle dependencies for different channels
|
|
err = rm.MarkDependenciesMet([]MessageID{msgID1}, channel1)
|
|
require.NoError(t, err)
|
|
|
|
err = rm.MarkDependenciesMet([]MessageID{msgID2}, channel2)
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
// Test that callbacks are correctly triggered for multiple channels
|
|
func TestMultiChannelCallbacks(t *testing.T) {
|
|
// rm1 is the manager we are testing callbacks on
|
|
rm1, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer rm1.Cleanup()
|
|
|
|
// rm2 simulates another peer
|
|
rm2, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer rm2.Cleanup()
|
|
|
|
channel1 := "callback-channel-1"
|
|
channel2 := "callback-channel-2"
|
|
|
|
var wg sync.WaitGroup
|
|
var cbMutex sync.Mutex
|
|
readyMessages := make(map[MessageID]string)
|
|
sentMessages := make(map[MessageID]string)
|
|
|
|
callbacks := EventCallbacks{
|
|
OnMessageReady: func(messageId MessageID, channelId string) {
|
|
// This will be triggered when rm1 receives messages from rm2
|
|
cbMutex.Lock()
|
|
readyMessages[messageId] = channelId
|
|
cbMutex.Unlock()
|
|
wg.Done()
|
|
},
|
|
OnMessageSent: func(messageId MessageID, channelId string) {
|
|
// This will be triggered when rm1's messages are acked by rm2
|
|
cbMutex.Lock()
|
|
sentMessages[messageId] = channelId
|
|
cbMutex.Unlock()
|
|
wg.Done()
|
|
},
|
|
}
|
|
rm1.RegisterCallbacks(callbacks)
|
|
|
|
// --- Test Scenario ---
|
|
// 1. rm1 sends one message on each channel.
|
|
msgID1_ch1 := MessageID("msg-on-ch1")
|
|
wrappedMsg1_ch1, err := rm1.WrapOutgoingMessage([]byte("payload1"), msgID1_ch1, channel1)
|
|
require.NoError(t, err)
|
|
|
|
msgID2_ch2 := MessageID("msg-on-ch2")
|
|
wrappedMsg2_ch2, err := rm1.WrapOutgoingMessage([]byte("payload2"), msgID2_ch2, channel2)
|
|
require.NoError(t, err)
|
|
|
|
// 2. rm2 receives these messages to update its history.
|
|
_, err = rm2.UnwrapReceivedMessage(wrappedMsg1_ch1)
|
|
require.NoError(t, err)
|
|
_, err = rm2.UnwrapReceivedMessage(wrappedMsg2_ch2)
|
|
require.NoError(t, err)
|
|
|
|
// 3. rm2 sends messages back on each channel, which will act as ACKs for rm1's messages.
|
|
ackID1_ch1 := MessageID("ack-on-ch1")
|
|
wrappedAck1_ch1, err := rm2.WrapOutgoingMessage([]byte("ack_payload1"), ackID1_ch1, channel1)
|
|
require.NoError(t, err)
|
|
|
|
ackID2_ch2 := MessageID("ack-on-ch2")
|
|
wrappedAck2_ch2, err := rm2.WrapOutgoingMessage([]byte("ack_payload2"), ackID2_ch2, channel2)
|
|
require.NoError(t, err)
|
|
|
|
// 4. rm1 receives the "ack" messages. This should trigger:
|
|
// - OnMessageSent for msgID1_ch1 and msgID2_ch2
|
|
// - OnMessageReady for ackID1_ch1 and ackID2_ch2
|
|
wg.Add(4) // Expect 2 sent and 2 ready callbacks
|
|
_, err = rm1.UnwrapReceivedMessage(wrappedAck1_ch1)
|
|
require.NoError(t, err)
|
|
_, err = rm1.UnwrapReceivedMessage(wrappedAck2_ch2)
|
|
require.NoError(t, err)
|
|
|
|
// --- Verification ---
|
|
waitTimeout(&wg, 5*time.Second, t)
|
|
|
|
cbMutex.Lock()
|
|
defer cbMutex.Unlock()
|
|
|
|
// Check that both original messages were marked as sent and on the correct channel
|
|
require.Equal(t, channel1, sentMessages[msgID1_ch1], "OnMessageSent for msg1 has incorrect channel")
|
|
require.Equal(t, channel2, sentMessages[msgID2_ch2], "OnMessageSent for msg2 has incorrect channel")
|
|
require.Len(t, sentMessages, 2, "Expected exactly 2 sent messages")
|
|
|
|
// Check that both ack messages were marked as ready and on the correct channel
|
|
require.Equal(t, channel1, readyMessages[ackID1_ch1], "OnMessageReady for ack1 has incorrect channel")
|
|
require.Equal(t, channel2, readyMessages[ackID2_ch2], "OnMessageReady for ack2 has incorrect channel")
|
|
require.Len(t, readyMessages, 2, "Expected exactly 2 ready messages")
|
|
}
|
|
|
|
// Test logging configuration functionality
|
|
func TestSetLogFile(t *testing.T) {
|
|
// Create a temporary directory for test logs
|
|
tempDir, err := os.MkdirTemp("", "nim-sds-test-logs")
|
|
require.NoError(t, err)
|
|
defer os.RemoveAll(tempDir)
|
|
|
|
// Test 1: Set a valid log file path
|
|
testLogFile := filepath.Join(tempDir, "nim-sds-test.log")
|
|
|
|
err = SetLogFile(testLogFile)
|
|
require.NoError(t, err)
|
|
|
|
// Test 2: Verify the environment variable was set correctly by creating a ReliabilityManager
|
|
// and checking that it doesn't return errors related to logging
|
|
rm, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer rm.Cleanup()
|
|
|
|
// Test 3: Try to generate some logs (this will trigger error logs internally)
|
|
testMessage := []byte("Test log message")
|
|
testMessageID := MessageID("test-log-msg")
|
|
testChannelID := "test-log-channel"
|
|
|
|
// This should work and potentially generate some logs
|
|
_, err = rm.WrapOutgoingMessage(testMessage, testMessageID, testChannelID)
|
|
if err != nil {
|
|
t.Logf("WrapOutgoingMessage failed (expected in some environments): %v", err)
|
|
}
|
|
|
|
// Give some time for potential async logging
|
|
time.Sleep(100 * time.Millisecond)
|
|
|
|
// Test 4: Check that log directory was created (if any logs were written)
|
|
if _, err := os.Stat(tempDir); os.IsNotExist(err) {
|
|
t.Log("Log directory was not created (no logs generated, which is fine)")
|
|
} else {
|
|
t.Log("Log directory exists (logs may have been generated)")
|
|
|
|
// Check if log file was created
|
|
if _, err := os.Stat(testLogFile); err == nil {
|
|
t.Log("Log file was created successfully")
|
|
|
|
// Read log file content to verify it's not empty
|
|
content, err := os.ReadFile(testLogFile)
|
|
if err != nil {
|
|
t.Logf("Could not read log file: %v", err)
|
|
} else {
|
|
t.Logf("Log file content length: %d bytes", len(content))
|
|
if len(content) > 0 {
|
|
t.Logf("Log file contains data: %s", string(content)[:min(100, len(content))])
|
|
}
|
|
}
|
|
} else {
|
|
t.Log("Log file was not created (no error logs generated)")
|
|
}
|
|
}
|
|
|
|
// Test 5: Test error case - empty log file path
|
|
err = SetLogFile("")
|
|
require.Error(t, err, "Expected error when setting empty log file path")
|
|
|
|
// Test 6: Test error case - nil path should work (it just sets empty env var)
|
|
err = SetLogFile("/tmp/test-path.log")
|
|
require.NoError(t, err, "Setting valid path should work")
|
|
}
|
|
|
|
// Test hybrid logging behavior by checking environment variable setting
|
|
func TestLogFileEnvironmentVariable(t *testing.T) {
|
|
// Save original environment
|
|
originalLogFile := os.Getenv("STATUS_SDS_LOG_FILE")
|
|
defer func() {
|
|
if originalLogFile != "" {
|
|
os.Setenv("STATUS_SDS_LOG_FILE", originalLogFile)
|
|
} else {
|
|
os.Unsetenv("STATUS_SDS_LOG_FILE")
|
|
}
|
|
}()
|
|
|
|
// Test that SetLogFile sets the environment variable correctly
|
|
testPath := "/test/path/nim-sds.log"
|
|
err := SetLogFile(testPath)
|
|
require.NoError(t, err)
|
|
|
|
// Check that environment variable was set
|
|
envValue := os.Getenv("STATUS_SDS_LOG_FILE")
|
|
require.Equal(t, testPath, envValue, "Environment variable STATUS_SDS_LOG_FILE should match set path")
|
|
|
|
// Test that we can create ReliabilityManager with custom log file
|
|
rm, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer rm.Cleanup()
|
|
|
|
t.Log("Successfully created ReliabilityManager with custom log file path")
|
|
}
|
|
|
|
// Test logging configuration with real message operations
|
|
func TestLoggingWithMessageOperations(t *testing.T) {
|
|
// Create temp directory
|
|
tempDir, err := os.MkdirTemp("", "nim-sds-msg-logs")
|
|
require.NoError(t, err)
|
|
defer os.RemoveAll(tempDir)
|
|
|
|
// Set log file
|
|
logFile := filepath.Join(tempDir, "message-ops.log")
|
|
err = SetLogFile(logFile)
|
|
require.NoError(t, err)
|
|
|
|
// Create RMs
|
|
senderRM, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer senderRM.Cleanup()
|
|
|
|
receiverRM, err := NewReliabilityManager()
|
|
require.NoError(t, err)
|
|
defer receiverRM.Cleanup()
|
|
|
|
// Perform operations that might generate logs
|
|
channelID := "logging-test-channel"
|
|
|
|
// Send messages
|
|
for i := 0; i < 5; i++ {
|
|
msgID := MessageID(fmt.Sprintf("log-test-msg-%d", i))
|
|
payload := []byte(fmt.Sprintf("Test message %d for logging", i))
|
|
|
|
wrappedMsg, err := senderRM.WrapOutgoingMessage(payload, msgID, channelID)
|
|
if err != nil {
|
|
t.Logf("WrapOutgoingMessage %d failed: %v", i, err)
|
|
continue
|
|
}
|
|
|
|
// Unwrap on receiver
|
|
_, err = receiverRM.UnwrapReceivedMessage(wrappedMsg)
|
|
if err != nil {
|
|
t.Logf("UnwrapReceivedMessage %d failed: %v", i, err)
|
|
}
|
|
}
|
|
|
|
// Wait for potential async logging
|
|
time.Sleep(200 * time.Millisecond)
|
|
|
|
// Check if logs were written
|
|
if _, err := os.Stat(logFile); err == nil {
|
|
content, err := os.ReadFile(logFile)
|
|
if err == nil && len(content) > 0 {
|
|
t.Logf("Log file created with %d bytes of content", len(content))
|
|
// Don't log full content as it might be large
|
|
} else {
|
|
t.Log("Log file exists but has no content")
|
|
}
|
|
} else {
|
|
t.Log("No log file was created (no errors generated)")
|
|
}
|
|
|
|
t.Log("Message operations completed successfully with logging configured")
|
|
}
|
|
|
|
// Helper function for min
|
|
func min(a, b int) int {
|
|
if a < b {
|
|
return a
|
|
}
|
|
return b
|
|
}
|