feat: Add timeout parameter to subscribers

README.md

@@ -16,15 +16,22 @@ go run .
 - eCAL 6 compatible (unreleased)
 - Pure cgo; no SWIG dependency
 - Custom C interface implementation
+- Direct deserialization from subscriber buffer to Go types
 
 Provides Go interfaces for:
-- [x] core
-- [x] publisher
-- [ ] subscriber
-- [ ] services
-- [ ] logging
-- [ ] monitoring
-- [ ] registration
+- [x] Core
+- [ ] Configuration
+- [x] Publisher
+  - [ ] Zero Copy
+- [x] Subscriber
+- [x] Message Types
+  - [x] Generic
+  - [x] String
+  - [x] Protobuf
+- [x] Logging
+- [ ] Services
+- [ ] Monitoring
+- [ ] Registration
 
 ## Non-system installations
 

ecal/protobuf/publisher/protobuf_publisher_test.go

@@ -1,23 +1,16 @@
-package publisher
+package publisher_test
 
 import (
 	"testing"
 
+	"github.com/DownerCase/ecal-go/internal/ecaltest/protobuf/testutil_publisher"
 	"github.com/DownerCase/ecal-go/protos"
 )
 
 func TestProtobufPublisher(t *testing.T) {
-	pub, err := New[protos.Person]()
-
-	if err != nil {
-		t.Error(err)
-	}
+	pub := testutil_publisher.NewProtobufPublisher[protos.Person](t, "testing_protobuf_publisher")
 	defer pub.Delete()
 
-	if err := pub.Create("testing"); err != nil {
-		t.Error(err)
-	}
-
 	if pub.Messages == nil {
 		t.Error("Message channel nil")
 	}

ecal/protobuf/subscriber/protobuf_subscriber.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"reflect"
+	"time"
 	"unsafe"
 
 	"github.com/DownerCase/ecal-go/ecal/subscriber"
@@ -32,9 +33,15 @@ func New[U any, T Msg[U]]() (*Subscriber[U, T], error) {
 	return psub, err
 }
 
-func (p *Subscriber[U, T]) Receive() (U, error) {
+func (p *Subscriber[U, T]) Receive(timeout time.Duration) (U, error) {
 	var u U
-	switch msg := (<-p.Messages).(type) {
+	var msg any
+	select {
+	case msg = <-p.Messages:
+	case <-time.After(timeout):
+		return u, errors.New("Receive timed out")
+	}
+	switch msg := msg.(type) {
 	case U:
 		return msg, nil
 	case error:

ecal/protobuf/subscriber/protobuf_subscriber_test.go

@@ -6,42 +6,35 @@ import (
 
 	"github.com/DownerCase/ecal-go/ecal"
 	"github.com/DownerCase/ecal-go/ecal/protobuf/publisher"
+	"github.com/DownerCase/ecal-go/internal/ecaltest"
+	"github.com/DownerCase/ecal-go/internal/ecaltest/protobuf/testutil_publisher"
 	"github.com/DownerCase/ecal-go/protos"
 )
 
-func TestSubscriber(t *testing.T) {
-	initResult := ecal.Initialize(
-		ecal.NewConfig(),
-		"Go eCAL!",
-		ecal.C_Publisher|ecal.C_Subscriber|ecal.C_Logging,
-	)
-	if initResult != 0 {
-		t.Fatal("Failed to initialize", initResult)
-	}
-	defer ecal.Finalize() // Shutdown eCAL at the end of the program
-
-	pub, err := publisher.New[protos.Person]()
+func newSubscriber[U any, T Msg[U]](t *testing.T, topic string) *Subscriber[U, T] {
+	sub, err := New[U, T]()
 	if err != nil {
 		t.Error(err)
 	}
-	defer pub.Delete()
-
-	if err := pub.Create("testing_string_subscriber"); err != nil {
+	if err := sub.Create(topic); err != nil {
 		t.Error(err)
 	}
+	return sub
+}
 
-	sub, err := New[protos.Person]()
-	if err != nil {
-		t.Error(err)
-	}
+func TestSubscriber(t *testing.T) {
+	ecaltest.InitEcal(t)
+	defer ecal.Finalize() // Shutdown eCAL at the end of the program
+
+	pub := testutil_publisher.NewProtobufPublisher[protos.Person](t, "testing_protobuf_subscriber")
+	defer pub.Delete()
+
+	sub := newSubscriber[protos.Person](t, "testing_protobuf_subscriber")
 	defer sub.Delete()
-	if err := sub.Create("testing_string_subscriber"); err != nil {
-		t.Error(err)
-	}
 
 	go sendMessages(pub)
 	for range 10 {
-		msg, err := sub.Receive()
+		msg, err := sub.Receive(2 * time.Second)
 
 		if err != nil {
 			t.Error(err)
@@ -61,6 +54,17 @@ func TestSubscriber(t *testing.T) {
 	}
 }
 
+func TestSubscriberTimeout(t *testing.T) {
+	ecaltest.InitEcal(t)
+	defer ecal.Finalize() // Shutdown eCAL at the end of the program
+	sub := newSubscriber[protos.Person](t, "testing_protobuf_subscriber_timeout")
+	defer sub.Delete()
+	msg, err := sub.Receive(50 * time.Millisecond)
+	if err == nil {
+		t.Error("Expected timeout, received message:", &msg)
+	}
+}
+
 func sendMessages(p *publisher.Publisher[*protos.Person]) {
 	person := &protos.Person{Id: 0, Name: "John", Email: "[email protected]",
 		Dog:   &protos.Dog{Name: "Pluto"},

ecal/publisher/publisher_test.go

@@ -1,12 +1,15 @@
-package publisher
+package publisher_test
 
 import (
 	"testing"
+
+	"github.com/DownerCase/ecal-go/ecal/publisher"
+	"github.com/DownerCase/ecal-go/internal/ecaltest/testutil_publisher"
 )
 
 func TestNewPublishers(t *testing.T) {
 	for range 100 {
-		ptr, err := New()
+		ptr, err := publisher.New()
 		if err != nil {
 			t.Error(err)
 		}
@@ -16,15 +19,8 @@ func TestNewPublishers(t *testing.T) {
 }
 
 func TestPublisher(t *testing.T) {
-	pub, err := New()
-	if err != nil {
-		t.Error(err)
-	}
+	pub := testutil_publisher.NewGenericPublisher(t, "testing")
 	defer pub.Delete()
-
-	if err := pub.Create("testing", DataType{}); err != nil {
-		t.Error(err)
-	}
 	if pub.Messages == nil {
 		t.Error("Message channel nil")
 	}

ecal/string/publisher/string_publisher_test.go

@@ -1,21 +1,15 @@
-package publisher
+package publisher_test
 
 import (
 	"testing"
-)
 
-func TestProtobufPublisher(t *testing.T) {
-	pub, err := New()
+	"github.com/DownerCase/ecal-go/internal/ecaltest/string/testutil_publisher"
+)
 
-	if err != nil {
-		t.Error(err)
-	}
+func TestStringPublisher(t *testing.T) {
+	pub := testutilpublisher.NewStringPublisher(t, "test_string_publisher")
 	defer pub.Delete()
 
-	if err := pub.Create("testing"); err != nil {
-		t.Error(err)
-	}
-
 	if pub.Messages == nil {
 		t.Error("Message channel nil")
 	}

ecal/string/subscriber/string_subscriber.go

@@ -2,6 +2,8 @@ package subscriber
 
 import "C"
 import (
+	"errors"
+	"time"
 	"unsafe"
 
 	"github.com/DownerCase/ecal-go/ecal/subscriber"
@@ -17,8 +19,13 @@ func New() (*Subscriber, error) {
 	return &Subscriber{*sub}, err
 }
 
-func (p *Subscriber) Receive() string {
-	return (<-p.Messages).(string)
+func (p *Subscriber) Receive(timeout time.Duration) (string, error) {
+	select {
+	case msg := <-p.Messages:
+		return msg.(string), nil
+	case <-time.After(timeout):
+		return "", errors.New("Receive timed out")
+	}
 }
 
 func deserialize(data unsafe.Pointer, len int) any {

ecal/string/subscriber/string_subscriber_test.go

@@ -7,43 +7,42 @@ import (
 
 	"github.com/DownerCase/ecal-go/ecal"
 	"github.com/DownerCase/ecal-go/ecal/string/publisher"
+	"github.com/DownerCase/ecal-go/internal/ecaltest"
+	"github.com/DownerCase/ecal-go/internal/ecaltest/string/testutil_publisher"
 )
 
 var TEST_MESSAGE = "Test string"
 
-func TestSubscriber(t *testing.T) {
-	initResult := ecal.Initialize(
-		ecal.NewConfig(),
-		"Go eCAL!",
-		ecal.C_Publisher|ecal.C_Subscriber|ecal.C_Logging,
-	)
-	if initResult != 0 {
-		t.Fatal("Failed to initialize", initResult)
-	}
-	defer ecal.Finalize() // Shutdown eCAL at the end of the program
-
-	pub, err := publisher.New()
+func newSubscriber(t *testing.T, topic string) *Subscriber {
+	sub, err := New()
 	if err != nil {
 		t.Error(err)
 	}
-	defer pub.Delete()
-
-	if err := pub.Create("testing_string_subscriber"); err != nil {
+	if err := sub.Create(topic); err != nil {
 		t.Error(err)
 	}
+	return sub
+}
 
-	sub, err := New()
-	if err != nil {
-		t.Error(err)
-	}
+// Export for testing
+var NewSubscriber = newSubscriber
+
+func TestSubscriber(t *testing.T) {
+	ecaltest.InitEcal(t)
+	defer ecal.Finalize() // Shutdown eCAL at the end of the program
+
+	pub := testutilpublisher.NewStringPublisher(t, "testing_string_subscriber")
+	defer pub.Delete()
+
+	sub := newSubscriber(t, "testing_string_subscriber")
 	defer sub.Delete()
-	if err := sub.Create("testing_string_subscriber"); err != nil {
-		t.Error(err)
-	}
 
 	go sendMessages(pub)
 	for range 10 {
-		msg := sub.Receive()
+		msg, err := sub.Receive(2 * time.Second)
+		if err != nil {
+			t.Error(err)
+		}
 		if len(msg) != len(TEST_MESSAGE) {
 			t.Error("Expected message of length", len(TEST_MESSAGE), "Received:", len(msg))
 		}
@@ -53,6 +52,17 @@ func TestSubscriber(t *testing.T) {
 	}
 }
 
+func TestSubscriberTimeout(t *testing.T) {
+	ecaltest.InitEcal(t)
+	defer ecal.Finalize() // Shutdown eCAL at the end of the program
+	sub := newSubscriber(t, "testing_string_subscriber_timeout")
+	defer sub.Delete()
+	msg, err := sub.Receive(50 * time.Millisecond)
+	if err == nil {
+		t.Error("Expected timeout, received message:", msg)
+	}
+}
+
 func sendMessages(p *publisher.Publisher) {
 	for !p.IsStopped() {
 		p.Messages <- []byte(TEST_MESSAGE)

ecal/subscriber/subscriber.go

@@ -13,6 +13,7 @@ import "C"
 import (
 	"errors"
 	"runtime/cgo"
+	"time"
 	"unsafe"
 
 	"github.com/DownerCase/ecal-go/ecal/msg"
@@ -74,8 +75,13 @@ func (p *Subscriber) Create(topic string, datatype DataType) error {
 }
 
 // Receive a new message from the eCAL receive callback
-func (p *Subscriber) Receive() []byte {
-	return (<-p.Messages).([]byte)
+func (p *Subscriber) Receive(timeout time.Duration) ([]byte, error) {
+	select {
+	case msg := <-p.Messages:
+		return msg.([]byte), nil
+	case <-time.After(timeout):
+		return nil, errors.New("Receive timed out")
+	}
 }
 
 // Deserialize straight from the eCAL internal buffer to our Go []byte