From a6162e2792c6c4a8ed41704cb85b5b97df4836cd Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 19:15:59 +0300 Subject: [PATCH 01/40] remove error pkg --- error.go | 20 -------------------- 1 file changed, 20 deletions(-) delete mode 100644 error.go diff --git a/error.go b/error.go deleted file mode 100644 index 2ae4837..0000000 --- a/error.go +++ /dev/null @@ -1,20 +0,0 @@ -package nats_pubsub_go - -import "errors" - -var ( - // ErrInvalidArgument is an error returned when an invalid argument is provided. - // This is used to indicate that a function or method has been called with - // arguments that do not meet the required criteria or format. - ErrInvalidArgument = errors.New("invalid argument") - - // ErrCloseConnection is an error returned when an operation is attempted - // on a closed connection. It signifies that the connection has been - // terminated and cannot be used for further operations. - ErrCloseConnection = errors.New("connection is close") - - // ErrConnectionAlreadyClosed indicates that a connection closure was attempted on an already closed connection. - // This error helps differentiate between the connection being in a valid state versus being redundantly closed. - // By defining this error, the code provides a specific signal to handle such redundant closure attempts gracefully. - ErrConnectionAlreadyClosed = errors.New("connection is already closed") -) From 17b9922a1547dd66849e8f1757ac9364a9510423 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 19:16:21 +0300 Subject: [PATCH 02/40] moved the errors into one package with interfaces --- pubsub.go | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/pubsub.go b/pubsub.go index 8cce628..dbdf896 100644 --- a/pubsub.go +++ b/pubsub.go @@ -1,11 +1,29 @@ package nats_pubsub_go import ( + "errors" "time" "github.com/nats-io/nats.go" ) +var ( + // ErrInvalidArgument is an error returned when an invalid argument is provided. + // This is used to indicate that a function or method has been called with + // arguments that do not meet the required criteria or format. + ErrInvalidArgument = errors.New("invalid argument") + + // ErrCloseConnection is an error returned when an operation is attempted + // on a closed connection. It signifies that the connection has been + // terminated and cannot be used for further operations. + ErrCloseConnection = errors.New("connection is close") + + // ErrConnectionAlreadyClosed indicates that a connection closure was attempted on an already closed connection. + // This error helps differentiate between the connection being in a valid state versus being redundantly closed. + // By defining this error, the code provides a specific signal to handle such redundant closure attempts gracefully. + ErrConnectionAlreadyClosed = errors.New("connection is already closed") +) + // Publisher defines the interface for a publisher that can send messages, // handle requests, and be closed. This interface abstracts the operations // that a concrete publisher implementation must provide. From 1acb3506687b3399a30818e31c6fc419b86575c5 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 19:17:23 +0300 Subject: [PATCH 03/40] init Subscriber --- pkg/subscriber.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/pkg/subscriber.go b/pkg/subscriber.go index 10e6d3e..7a2b485 100644 --- a/pkg/subscriber.go +++ b/pkg/subscriber.go @@ -1,6 +1,7 @@ package pkg import ( + "context" pubsub "github.com/SyntaxErrorLineNULL/nats-pubsub-go" "github.com/nats-io/nats.go" "sync/atomic" @@ -28,6 +29,10 @@ func NewSubscriber(conn *nats.Conn) *Subscriber { return &Subscriber{conn: conn} } +func (s *Subscriber) Subscriber(ctx context.Context, subject string) (<-chan pubsub.MessageHandler, error) { + return nil, nil +} + // AsyncSubscribe subscribes to a subject asynchronously and returns a Subscription object // that allows receiving messages on the subscribed subject. It handles errors and provides // a channel for receiving messages. From c660adc6d7e1fce6504968cc0474a7d77b65104f Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 19:29:07 +0300 Subject: [PATCH 04/40] message init --- message.go | 1 + 1 file changed, 1 insertion(+) create mode 100644 message.go diff --git a/message.go b/message.go new file mode 100644 index 0000000..5b99b3b --- /dev/null +++ b/message.go @@ -0,0 +1 @@ +package nats_pubsub_go From 9d89b232e605ae625ca201b9f2fa24b38884b765 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 19:29:36 +0300 Subject: [PATCH 05/40] message struct --- message.go | 36 ++++++++++++++++++++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/message.go b/message.go index 5b99b3b..73ae463 100644 --- a/message.go +++ b/message.go @@ -1 +1,37 @@ package nats_pubsub_go + +import ( + "context" + "github.com/nats-io/nats.go" + "sync" +) + +type Container []byte + +type Header map[string][]string + +type Message struct { + RequestID string + Container Container + + // Message is a NATS message data. + // This channel allows consumers to process incoming messages + // that are published to the NATS subject this handler is subscribed to. + message *nats.Msg + + // Subscription is the underlying NATS subscription. + // This represents the subscription to a NATS subject or subjects, + // allowing the handler to receive messages from NATS. + subscription *nats.Subscription + + // once is used to ensure certain operations are performed only once. + // It uses sync.Once to guarantee that specific actions, such as closing + // the channel and unsubscribing, are executed only a single time. + once sync.Once + + parentCtx context.Context +} + +func NewMessage() *Message { + return &Message{} +} From 24022f25365a15fcff33152d42d7abb5e372f991 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 19:34:00 +0300 Subject: [PATCH 06/40] add descriptions --- message.go | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/message.go b/message.go index 73ae463..4521e63 100644 --- a/message.go +++ b/message.go @@ -6,12 +6,24 @@ import ( "sync" ) +// Container represents a byte slice used to store data within the Message structure. +// It acts as a container for the payload associated with the message. type Container []byte +// Header represents a map of string keys to slices of string values. +// It is used to store metadata or headers associated with a message. type Header map[string][]string +// Message represents a communication unit in a NATS-based messaging system. +// It encapsulates the necessary components for processing messages, +// including the payload, metadata, and underlying NATS-specific details. type Message struct { + // RequestID is a unique identifier for the message. + // This ID is used to track and correlate requests and responses in the messaging system. RequestID string + + // Container holds the payload of the message as a byte slice. + // It represents the data being transmitted or processed in the communication. Container Container // Message is a NATS message data. @@ -29,9 +41,11 @@ type Message struct { // the channel and unsubscribing, are executed only a single time. once sync.Once + // parentCtx is the context associated with the message's parent operation. + // It provides a way to propagate cancellation, timeouts, or deadlines across operations. parentCtx context.Context } -func NewMessage() *Message { - return &Message{} +func NewMessage(parentCtx context.Context) *Message { + return &Message{parentCtx: parentCtx} } From c38dea9b8d663975f5adda4a0eb6479d2b8fd9aa Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 19:42:19 +0300 Subject: [PATCH 07/40] GetContainer method --- message.go | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/message.go b/message.go index 4521e63..f2c6b3d 100644 --- a/message.go +++ b/message.go @@ -49,3 +49,12 @@ type Message struct { func NewMessage(parentCtx context.Context) *Message { return &Message{parentCtx: parentCtx} } + +// GetContainer retrieves the container payload from the underlying NATS message data. +// This method returns the raw data associated with the message, allowing consumers +// to access the payload for further processing or handling. +func (msg *Message) GetContainer() Container { + // Access and return the data field from the underlying NATS message. + // This represents the payload of the message that was received from NATS. + return msg.message.Data +} From 61d62664e5c79c10e6ad09d5f17f9cda130295ad Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 19:42:40 +0300 Subject: [PATCH 08/40] GetHeader method --- message.go | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/message.go b/message.go index f2c6b3d..9900e9b 100644 --- a/message.go +++ b/message.go @@ -58,3 +58,12 @@ func (msg *Message) GetContainer() Container { // This represents the payload of the message that was received from NATS. return msg.message.Data } + +// GetHeader retrieves the header information from the underlying NATS message. +// This method converts the header from the NATS message into the custom Header type, +// allowing consumers to access metadata associated with the message in a structured manner. +func (msg *Message) GetHeader() Header { + // Access and convert the header field from the underlying NATS message. + // The header contains key-value pairs representing metadata about the message. + return Header(msg.message.Header) +} From b5d215c1c42b7b39d3ea88ee3696221b494b6c0a Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 19:43:29 +0300 Subject: [PATCH 09/40] Ack method --- message.go | 39 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/message.go b/message.go index 9900e9b..806d78d 100644 --- a/message.go +++ b/message.go @@ -4,6 +4,7 @@ import ( "context" "github.com/nats-io/nats.go" "sync" + "time" ) // Container represents a byte slice used to store data within the Message structure. @@ -67,3 +68,41 @@ func (msg *Message) GetHeader() Header { // The header contains key-value pairs representing metadata about the message. return Header(msg.message.Header) } + +// Ack acknowledges the receipt of the message, notifying the NATS system that it has been processed. +// It supports an optional timeout for acknowledgment and cancels if the parent context is done. +// If a timeout is not specified, the message is acknowledged immediately. +// If the parent context is canceled before the timeout, the acknowledgment is aborted, and the context error is returned. +func (msg *Message) Ack(timeout time.Duration) error { + // Check if no timeout is specified. + // If timeout is zero, acknowledge the message immediately without delay. + if timeout == 0 { + // Acknowledge the message immediately when no timeout is set. + return msg.message.Ack() + } + + // Create a ticker that will emit an event after the specified timeout duration. + // This provides a mechanism to handle the delay in acknowledging the message, based on the given timeout. + ticker := time.NewTicker(timeout) + // Ensure that the ticker is properly stopped after use to release any associated resources. + // This is important to avoid potential resource leaks or unnecessary background work. + defer ticker.Stop() + + // Use a select statement to wait for either the timeout or a cancellation signal from the parent context. + // This allows the function to handle both the timeout event and the context cancellation in a non-blocking manner. + select { + // Case for when the ticker triggers, signaling the timeout has elapsed. + // Acknowledge the message at this point. + case <-ticker.C: + // Acknowledge the message after the timeout has elapsed. + // This ensures that the message is acknowledged only after the specified waiting period. + return msg.message.Ack() + + // Case for when the parent context is canceled before the timeout. + // Return the context's error to indicate that the acknowledgment was not completed. + case <-msg.parentCtx.Done(): + // Return the error associated with the context cancellation. + // This ensures that the operation is properly terminated if the context is canceled. + return msg.parentCtx.Err() + } +} From 3b8dbb72f2bff449a2d93447abbbe5eaa4ab5b1c Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 19:43:49 +0300 Subject: [PATCH 10/40] Nak method --- message.go | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/message.go b/message.go index 806d78d..08b477e 100644 --- a/message.go +++ b/message.go @@ -106,3 +106,20 @@ func (msg *Message) Ack(timeout time.Duration) error { return msg.parentCtx.Err() } } + +// Nak sends a negative acknowledgment for the message, signaling that it was not processed successfully. +// This can inform the NATS system to requeue the message for further processing or take alternative action. +// It provides support for an optional delay before sending the negative acknowledgment. +// If a timeout is specified, it uses NakWithDelay to apply the delay; otherwise, it sends an immediate Nak. +func (msg *Message) Nak(timeout time.Duration) error { + // Check if no timeout is specified for the negative acknowledgment. + // If timeout is zero, immediately send the negative acknowledgment without any delay. + if timeout == 0 { + // Directly send a negative acknowledgment for the message, indicating immediate rejection. + return msg.message.Nak() + } + + // Send a negative acknowledgment with the specified delay. + // This uses the provided timeout to delay the rejection, which can be useful in specific scenarios. + return msg.message.NakWithDelay(timeout) +} From 04ec190743d50cf0cdd00886d346227e2c4026f5 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 19:44:05 +0300 Subject: [PATCH 11/40] Respond method --- message.go | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/message.go b/message.go index 08b477e..1e2e1ae 100644 --- a/message.go +++ b/message.go @@ -123,3 +123,12 @@ func (msg *Message) Nak(timeout time.Duration) error { // This uses the provided timeout to delay the rejection, which can be useful in specific scenarios. return msg.message.NakWithDelay(timeout) } + +// Respond sends a response back to the sender of the message. +// This is typically used in a request-response pattern where the sender expects a reply to the message it sent. +// The data parameter contains the payload to be sent as the response. +func (msg *Message) Respond(data []byte) error { + // Use the NATS Respond method to send the provided data as a response to the message. + // This operation communicates the reply to the message's sender, adhering to the NATS messaging protocol. + return msg.message.Respond(data) +} From b41d0db78084e7c9ac8b8d5ef52a14ac2613f48a Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 19:44:25 +0300 Subject: [PATCH 12/40] fix import --- message.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/message.go b/message.go index 1e2e1ae..1af7c7f 100644 --- a/message.go +++ b/message.go @@ -2,9 +2,10 @@ package nats_pubsub_go import ( "context" - "github.com/nats-io/nats.go" "sync" "time" + + "github.com/nats-io/nats.go" ) // Container represents a byte slice used to store data within the Message structure. From 4f04fc8e2311d60df2f5f100eb7b5de8cd5c39f6 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 21:11:40 +0300 Subject: [PATCH 13/40] receive message --- message.go | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/message.go b/message.go index 1af7c7f..eef73ae 100644 --- a/message.go +++ b/message.go @@ -52,6 +52,22 @@ func NewMessage(parentCtx context.Context) *Message { return &Message{parentCtx: parentCtx} } +// ReceiveMessage waits for the next message on the subscription with the specified timeout duration. +// It returns the received NATS message or an error if the operation times out. +// Note: This method can only be used with SyncSubscribe. +func (msg *Message) ReceiveMessage(timeout time.Duration) (*nats.Msg, error) { + // Wait for the next message on the subscription with the given timeout duration. + // The NextMsg method blocks until a message is received or the timeout is reached. + // If a message is received, it is returned; otherwise, an error is returned. + nextMessage, err := msg.subscription.NextMsg(timeout) + if err != nil { + return nil, err + } + // Return the received message along with a nil error if NextMsg succeeds. + // This means the message was successfully retrieved within the specified timeout. + return nextMessage, nil +} + // GetContainer retrieves the container payload from the underlying NATS message data. // This method returns the raw data associated with the message, allowing consumers // to access the payload for further processing or handling. From 9e3765a52c4b6f2b49d9d75ce5e813c81d9087d0 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 21:14:24 +0300 Subject: [PATCH 14/40] unsubscribe --- message.go | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/message.go b/message.go index eef73ae..f378e1d 100644 --- a/message.go +++ b/message.go @@ -68,6 +68,25 @@ func (msg *Message) ReceiveMessage(timeout time.Duration) (*nats.Msg, error) { return nextMessage, nil } +// Unsubscribe terminates the subscription and closes the data channel. +// It ensures that the channel is closed only once and that the subscription +// is properly unsubscribed from. This method helps clean up resources +// and prevent memory leaks or dangling subscriptions. +func (msg *Message) Unsubscribe() (err error) { + // Ensure the Data channel is closed only once by using the sync.Once mechanism. + // The sync.Once type ensures that the provided function is executed only once, + // regardless of how many times it's called. + msg.once.Do(func() { + + // Unsubscribe from the current subscription to stop receiving messages. + // The Unsubscribe method call removes the subscription and cleans up resources. + err = msg.subscription.Unsubscribe() + }) + + // Return any error encountered during the Unsubscribe process. + return err +} + // GetContainer retrieves the container payload from the underlying NATS message data. // This method returns the raw data associated with the message, allowing consumers // to access the payload for further processing or handling. From 9b59b59877c18feefd16ec448aafb28d9ad40210 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 21:16:18 +0300 Subject: [PATCH 15/40] check if subscriber is close --- pkg/subscriber.go | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/pkg/subscriber.go b/pkg/subscriber.go index 7a2b485..e2a8727 100644 --- a/pkg/subscriber.go +++ b/pkg/subscriber.go @@ -29,7 +29,14 @@ func NewSubscriber(conn *nats.Conn) *Subscriber { return &Subscriber{conn: conn} } -func (s *Subscriber) Subscriber(ctx context.Context, subject string) (<-chan pubsub.MessageHandler, error) { +func (s *Subscriber) Subscriber(ctx context.Context, subject, queue string) (<-chan pubsub.MessageHandler, error) { + // Check if the subscriber is closed. If closed, return an ErrCloseConnection error. + // This prevents a situation where the client has closed the Subscriber but then tries to perform some manipulations afterwards, guaranteeing + // that no operations will be performed on a closed instance. + if s.isClose.Load() { + return nil, pubsub.ErrCloseConnection + } + return nil, nil } From d20a649228303d08587d7f81706537114d304f22 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 21:17:01 +0300 Subject: [PATCH 16/40] check subject and queue is empty --- pkg/subscriber.go | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/pkg/subscriber.go b/pkg/subscriber.go index e2a8727..9a623db 100644 --- a/pkg/subscriber.go +++ b/pkg/subscriber.go @@ -37,6 +37,13 @@ func (s *Subscriber) Subscriber(ctx context.Context, subject, queue string) (<-c return nil, pubsub.ErrCloseConnection } + // Check if the provided subject or queue is empty. + // An empty subject or queue is invalid and cannot be subscribed to. + // Return an ErrInvalidArgument error to indicate the issue. + if subject == "" || queue == "" { + return nil, pubsub.ErrInvalidArgument + } + return nil, nil } From bc9d1ba515cc392a36988c7bd2aef5f72f43a72e Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 21:34:18 +0300 Subject: [PATCH 17/40] init encode (message serialization) --- pkg/encode.go | 1 + 1 file changed, 1 insertion(+) create mode 100644 pkg/encode.go diff --git a/pkg/encode.go b/pkg/encode.go new file mode 100644 index 0000000..c1caffe --- /dev/null +++ b/pkg/encode.go @@ -0,0 +1 @@ +package pkg From 030b8703041083baf387eef383f8b3ae14e35c7e Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 21:36:57 +0300 Subject: [PATCH 18/40] encode interface --- pkg/encode.go | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/pkg/encode.go b/pkg/encode.go index c1caffe..342b2b3 100644 --- a/pkg/encode.go +++ b/pkg/encode.go @@ -1 +1,7 @@ package pkg + +import "github.com/nats-io/nats.go" + +type Encoder interface { + Encode(*nats.Msg) (*Message, error) +} From 7823984fa0a17ddeec01edee6c416e2cc579ed51 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 21:42:20 +0300 Subject: [PATCH 19/40] add json serialization lib --- go.mod | 1 + go.sum | 2 ++ 2 files changed, 3 insertions(+) diff --git a/go.mod b/go.mod index 1918b42..2da957b 100644 --- a/go.mod +++ b/go.mod @@ -16,6 +16,7 @@ require ( github.com/nats-io/nkeys v0.4.7 // indirect github.com/nats-io/nuid v1.0.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/segmentio/encoding v0.4.1 // indirect golang.org/x/crypto v0.18.0 // indirect golang.org/x/sys v0.16.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/go.sum b/go.sum index aabedab..d4939a2 100644 --- a/go.sum +++ b/go.sum @@ -16,6 +16,8 @@ github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/segmentio/encoding v0.4.1 h1:KLGaLSW0jrmhB58Nn4+98spfvPvmo4Ci1P/WIQ9wn7w= +github.com/segmentio/encoding v0.4.1/go.mod h1:/d03Cd8PoaDeceuhUUUQWjU0KhWjrmYrWPgtJHYZSnI= github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= golang.org/x/crypto v0.18.0 h1:PGVlW0xEltQnzFZ55hkuX5+KLyrMYhHld1YHO4AKcdc= From 669edb5953bb3d5e96155e0f2ca22ab06a69c020 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 21:55:38 +0300 Subject: [PATCH 20/40] add request time, and add json fields --- message.go => pkg/message.go | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) rename message.go => pkg/message.go (95%) diff --git a/message.go b/pkg/message.go similarity index 95% rename from message.go rename to pkg/message.go index f378e1d..a14a853 100644 --- a/message.go +++ b/pkg/message.go @@ -1,4 +1,4 @@ -package nats_pubsub_go +package pkg import ( "context" @@ -22,11 +22,15 @@ type Header map[string][]string type Message struct { // RequestID is a unique identifier for the message. // This ID is used to track and correlate requests and responses in the messaging system. - RequestID string + RequestID string `json:"request_id,omitempty"` // Container holds the payload of the message as a byte slice. // It represents the data being transmitted or processed in the communication. - Container Container + Container Container `json:"container,omitempty"` + + // RequestTime records the time when the request was created or received. + // This timestamp is valuable for tracking message lifecycle and processing logic of some tasks. + RequestTime time.Time `json:"request_time,omitempty"` // Message is a NATS message data. // This channel allows consumers to process incoming messages From d6a1c41dab86e1821d53d8f06c37a2cf1eae5db7 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 22:03:46 +0300 Subject: [PATCH 21/40] Subject and Header fields --- pkg/message.go | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/pkg/message.go b/pkg/message.go index a14a853..a75b32f 100644 --- a/pkg/message.go +++ b/pkg/message.go @@ -20,6 +20,15 @@ type Header map[string][]string // It encapsulates the necessary components for processing messages, // including the payload, metadata, and underlying NATS-specific details. type Message struct { + // Subject defines the NATS subject associated with this message. + // It serves as the primary routing key for message delivery within the NATS system. + Subject string + + // Header contains metadata associated with the message in key-value format. + // It provides additional context or configuration for the message, + // allowing consumers to interpret or process it effectively. + Header Header + // RequestID is a unique identifier for the message. // This ID is used to track and correlate requests and responses in the messaging system. RequestID string `json:"request_id,omitempty"` From 5dbcce9b311ac848210d99a70fda3932e367da0d Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 23:16:02 +0300 Subject: [PATCH 22/40] encode test init --- pkg/encode.go | 46 +++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 43 insertions(+), 3 deletions(-) diff --git a/pkg/encode.go b/pkg/encode.go index 342b2b3..d7ff70a 100644 --- a/pkg/encode.go +++ b/pkg/encode.go @@ -1,7 +1,47 @@ package pkg -import "github.com/nats-io/nats.go" +import ( + "bytes" + "github.com/nats-io/nats.go" + "github.com/segmentio/encoding/json" +) -type Encoder interface { - Encode(*nats.Msg) (*Message, error) +// Decoder defines an interface for decoding a custom Message into a NATS message. +// It abstracts the process of converting application-level message representations +// into the NATS message format to facilitate communication within a NATS-based system. +type Decoder interface { + // Decode is responsible for transforming a custom Message into a NATS message. + // The resulting NATS message contains serialized data and metadata, formatted + // for compatibility with the NATS messaging system. + // Returns the NATS message if successful or an error if the transformation fails. + Decode(msg *Message) (*nats.Msg, error) +} + +// Encoding implements the Decoder interface, providing functionality to convert +// a custom Message structure into a format compatible with the NATS messaging system. +// It uses JSON serialization to encode the Message and populate the fields of a NATS message. +type Encoding struct{} + +// Decode converts a custom Message into a NATS message. +// This method serializes the Message's content, including its payload and metadata, +// into JSON format and assigns the serialized data to the NATS message fields. +func (Encoding) Decode(msg *Message) (*nats.Msg, error) { + // Initializes a buffer to temporarily hold the JSON-encoded data. + var buffer bytes.Buffer + + // Creates a JSON encoder that writes to the buffer. + // The encoder is responsible for serializing the Message into JSON format. + encoder := json.NewEncoder(&buffer) + + // Attempts to encode the custom Message into JSON and store it in the buffer. + // If an error occurs during encoding, the function immediately returns the error. + if err := encoder.Encode(msg); err != nil { + // Returns an error indicating that the encoding process failed. + return nil, err + } + + // Constructs a NATS message using the encoded data and message metadata. + // The Subject, Data, and Header fields of the NATS message are populated + // using the corresponding fields from the custom Message. + return &nats.Msg{Subject: msg.Subject, Data: buffer.Bytes(), Header: nats.Header(msg.Header)}, nil } From b008b46b525779158189a5c1ce0d97719fad8430 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Tue, 10 Dec 2024 23:29:10 +0300 Subject: [PATCH 23/40] encode test init --- pkg/encode_test.go | 1 + 1 file changed, 1 insertion(+) create mode 100644 pkg/encode_test.go diff --git a/pkg/encode_test.go b/pkg/encode_test.go new file mode 100644 index 0000000..c1caffe --- /dev/null +++ b/pkg/encode_test.go @@ -0,0 +1 @@ +package pkg From 44046ebbe6b79913b8703db0812ecb810352c1a2 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Wed, 11 Dec 2024 00:44:19 +0300 Subject: [PATCH 24/40] validate message method --- pkg/message.go | 37 +++++++++++++++++++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/pkg/message.go b/pkg/message.go index a75b32f..4ab03d5 100644 --- a/pkg/message.go +++ b/pkg/message.go @@ -5,6 +5,7 @@ import ( "sync" "time" + pubsub "github.com/SyntaxErrorLineNULL/nats-pubsub-go" "github.com/nats-io/nats.go" ) @@ -65,6 +66,42 @@ func NewMessage(parentCtx context.Context) *Message { return &Message{parentCtx: parentCtx} } +// Validate checks the integrity and validity of a Message object. +// It ensures that required fields are populated and that the headers contain valid data. +// If any validation rule fails, it returns an appropriate error. +func (msg *Message) Validate() error { + // Uses a switch statement to evaluate multiple conditions for validation. + // Each case addresses a specific aspect of the Message object's structure. + switch { + // Checks if the Subject field is empty. + // The Subject is a mandatory field, so an empty value is considered invalid. + case msg.Subject == "": + // Returns an error indicating the argument is invalid. + return pubsub.ErrInvalidArgument + + // Checks if the Header field is not nil, ensuring headers are valid. + // Iterates through each key-value pair in the Header map to validate its content. + case msg.Header != nil: + for key, values := range msg.Header { + // Validates that the key is non-empty and that the values slice has at least one element. + // An empty key or an empty values slice is considered invalid. + if key == "" || len(values) == 0 { + // Returns an error if the header data is invalid. + return pubsub.ErrInvalidArgument + } + } + + // Checks if the Container field is empty. + // The Container field must contain data for the Message to be valid. + case len(msg.Container) == 0: + // Returns an error indicating the Container is invalid. + return pubsub.ErrInvalidArgument + } + + // If all validation checks pass, returns nil to indicate the Message object is valid. + return nil +} + // ReceiveMessage waits for the next message on the subscription with the specified timeout duration. // It returns the received NATS message or an error if the operation times out. // Note: This method can only be used with SyncSubscribe. From 08b1af39eff11cea62ee372ca2c1bfa39e1fca51 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Wed, 11 Dec 2024 00:44:47 +0300 Subject: [PATCH 25/40] check if message is empty and use valid message method --- pkg/encode.go | 22 ++++++++++++++++++---- 1 file changed, 18 insertions(+), 4 deletions(-) diff --git a/pkg/encode.go b/pkg/encode.go index d7ff70a..2075100 100644 --- a/pkg/encode.go +++ b/pkg/encode.go @@ -2,6 +2,8 @@ package pkg import ( "bytes" + "errors" + "github.com/nats-io/nats.go" "github.com/segmentio/encoding/json" ) @@ -26,13 +28,25 @@ type Encoding struct{} // This method serializes the Message's content, including its payload and metadata, // into JSON format and assigns the serialized data to the NATS message fields. func (Encoding) Decode(msg *Message) (*nats.Msg, error) { - // Initializes a buffer to temporarily hold the JSON-encoded data. - var buffer bytes.Buffer + // Checks if the provided Message object is nil. + // If it is nil, an error is returned immediately, indicating the message is empty. + if msg == nil { + return nil, errors.New("message is empty") + } + + // Validates the provided Message object using its Validate method. + // If validation fails, the error is returned, indicating what went wrong. + if err := msg.Validate(); err != nil { + return nil, err + } + + // Initializes a new buffer that will temporarily hold the JSON-encoded data. + // This buffer acts as the target for the encoding process. + buffer := new(bytes.Buffer) // Creates a JSON encoder that writes to the buffer. // The encoder is responsible for serializing the Message into JSON format. - encoder := json.NewEncoder(&buffer) - + encoder := json.NewEncoder(buffer) // Attempts to encode the custom Message into JSON and store it in the buffer. // If an error occurs during encoding, the function immediately returns the error. if err := encoder.Encode(msg); err != nil { From 9d282d5f738f299af56cd70f025dcf7d8e59a2c1 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Wed, 11 Dec 2024 00:45:19 +0300 Subject: [PATCH 26/40] get pkg for segmentio --- go.mod | 3 ++- go.sum | 2 ++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/go.mod b/go.mod index 2da957b..abc32ca 100644 --- a/go.mod +++ b/go.mod @@ -6,6 +6,7 @@ require ( github.com/nats-io/gnatsd v1.4.1 github.com/nats-io/nats-server v1.4.1 github.com/nats-io/nats.go v1.37.0 + github.com/segmentio/encoding v0.4.1 github.com/stretchr/testify v1.9.0 ) @@ -16,7 +17,7 @@ require ( github.com/nats-io/nkeys v0.4.7 // indirect github.com/nats-io/nuid v1.0.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect - github.com/segmentio/encoding v0.4.1 // indirect + github.com/segmentio/asm v1.1.3 // indirect golang.org/x/crypto v0.18.0 // indirect golang.org/x/sys v0.16.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/go.sum b/go.sum index d4939a2..528831e 100644 --- a/go.sum +++ b/go.sum @@ -16,6 +16,8 @@ github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/segmentio/asm v1.1.3 h1:WM03sfUOENvvKexOLp+pCqgb/WDjsi7EK8gIsICtzhc= +github.com/segmentio/asm v1.1.3/go.mod h1:Ld3L4ZXGNcSLRg4JBsZ3//1+f/TjYl0Mzen/DQy1EJg= github.com/segmentio/encoding v0.4.1 h1:KLGaLSW0jrmhB58Nn4+98spfvPvmo4Ci1P/WIQ9wn7w= github.com/segmentio/encoding v0.4.1/go.mod h1:/d03Cd8PoaDeceuhUUUQWjU0KhWjrmYrWPgtJHYZSnI= github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= From 872228f991d56519bf16885813d6af5c5e102694 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Wed, 11 Dec 2024 00:45:37 +0300 Subject: [PATCH 27/40] encode test cases(test decode) --- pkg/encode_test.go | 93 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 93 insertions(+) diff --git a/pkg/encode_test.go b/pkg/encode_test.go index c1caffe..68f4d82 100644 --- a/pkg/encode_test.go +++ b/pkg/encode_test.go @@ -1 +1,94 @@ package pkg + +import ( + "bytes" + "fmt" + "testing" + + "github.com/nats-io/nats.go" + "github.com/segmentio/encoding/json" + "github.com/stretchr/testify/assert" +) + +func TestEncodingDecode(t *testing.T) { + // Test the Decode function of the Encoding type for different message scenarios. + // This test case is designed to validate that the Decode function correctly handles various + // cases, including valid messages, nil messages, and messages with invalid headers. + // The tests check for correct error handling, message transformation, and matching expected output. + cases := []struct { + name string + input *Message + expectError bool + expectedMsg *nats.Msg + }{ + {name: "Valid Message", input: &Message{Subject: "test.subject", Header: Header{"key": {"value1", "value2"}}, RequestID: "12345", Container: []byte("test payload")}, expectError: false, expectedMsg: &nats.Msg{Subject: "test.subject", Header: nats.Header{"key": []string{"value1", "value2"}}, Data: json.RawMessage(`{"subject":"test.subject","header":{"key":["value1","value2"]},"request_id":"12345","container":"dGVzdCBwYXlsb2Fk"}`)}}, + {name: "Nil Message", input: nil, expectError: true, expectedMsg: nil}, + {name: "Invalid Header", input: &Message{Subject: "test.subject", Header: Header{"key": nil}, RequestID: "12345", Container: []byte("test payload")}, expectError: true, expectedMsg: nil}, + {name: "Empty Header", input: &Message{Subject: "test.subject", RequestID: "12345", Container: []byte("test payload")}, expectError: false, expectedMsg: &nats.Msg{Subject: "test.subject", Data: json.RawMessage(`{"subject":"test.subject","request_id":"12345","container":"dGVzdCBwYXlsb2Fk"}`)}}, + } + + // Declare a variable of type Encoding. This variable will hold the instance of the encoding/decoding functionality + // that will be tested in the test cases. + var encoder Encoding + + // Iterate over the slice of test cases, 'cases', to execute the test logic for each test case individually. + // This loop allows us to run the same set of assertions on different inputs and expected outputs, ensuring + // that the function works across a variety of scenarios. + for _, tt := range cases { + // Start a subtest for the current test case, using 'tt.name' as the name of the subtest. + // This ensures that each test case is isolated, and we can easily identify which test case failed + // if any errors occur. The subtest allows for independent results and better debugging. + t.Run(tt.name, func(t *testing.T) { + // Call the 'Decode' method of the encoder, passing the 'input' from the current test case (tt.input). + // This is the core functionality being tested, where the input data is processed by the decoder. + // The result of the decoding operation is stored in the 'result' variable, and any errors that occur + // are captured in the 'err' variable. + result, err := encoder.Decode(tt.input) + + // Check if the current test case expects an error. The 'expectError' field in the test case struct + // indicates whether an error is expected during the 'Decode' function call. If 'expectError' is true, + // the assertions will verify that the function behaves as expected by producing an error. + if tt.expectError { + // Assert that an error has occurred during the decoding process. This assertion ensures that the error + // returned from 'Decode' is not nil when the test case expects an error. If no error occurs, the test will fail. + assert.Error(t, err, "Expected an error but got none for test case: "+tt.name) + // Assert that the 'result' is nil when an error is expected. This checks that the decoder does not + // return a valid result when an error occurs. If a valid result is returned when an error was expected, + // the test will fail. + assert.Nil(t, result, "Expected nil result but got a valid result for test case: "+tt.name) + } else { + // Assert that no error occurred during the decoding process. This check ensures that the function behaves as expected + // when an error is not anticipated. If an error is returned when it shouldn't be, the test will fail. + assert.NoError(t, err, fmt.Sprintf("Expected no error but got one for test case: %s", tt.name)) + + // Assert that the 'result' is not nil when no error is expected. This checks that a valid decoded result + // is returned by the 'Decode' function. If the result is nil, the test will fail. + assert.NotNil(t, result, fmt.Sprintf("Expected a valid result but got nil for test case: %s", tt.name)) + + // Assert that the 'Subject' field of the result matches the expected value from the test case. + // This checks that the 'Decode' function correctly transfers the 'Subject' from the input to the output. + assert.Equal(t, tt.expectedMsg.Subject, result.Subject, fmt.Sprintf("Subject mismatch for test case: %s", tt.name)) + + // Assert that the 'Header' field of the result matches the expected value from the test case. + // This ensures that the 'Header' is properly decoded and matches the expected format. + assert.Equal(t, tt.expectedMsg.Header, result.Header, fmt.Sprintf("Header mismatch for test case: %s", tt.name)) + + // Create a buffer to hold the expected JSON-encoded data. This buffer will store the result of encoding the input + // message into a JSON format for comparison against the 'Data' field of the result. + var expectedData bytes.Buffer + + // Encode the input message into the 'expectedData' buffer. This step ensures that the expected JSON data + // is correctly generated before comparing it with the actual 'Data' field in the result. + err = json.NewEncoder(&expectedData).Encode(tt.input) + + // Assert that no error occurred during the encoding process. If an error is returned, it will indicate a + // problem with the encoding step, causing the test to fail. + assert.NoError(t, err, fmt.Sprintf("Expected no encoding error but got one for test case: %s", tt.name)) + + // Assert that the 'Data' field of the result matches the expected JSON data generated above. + // This verifies that the 'Decode' function correctly processes the input and generates the expected data. + assert.Equal(t, expectedData.Bytes(), result.Data, fmt.Sprintf("Data mismatch for test case: %s", tt.name)) + } + }) + } +} From 1ee37f89e56673b1282511285d9ffd5a17b6e989 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Wed, 11 Dec 2024 01:00:07 +0300 Subject: [PATCH 28/40] update constructor --- pkg/message.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/message.go b/pkg/message.go index 4ab03d5..02ce975 100644 --- a/pkg/message.go +++ b/pkg/message.go @@ -62,8 +62,8 @@ type Message struct { parentCtx context.Context } -func NewMessage(parentCtx context.Context) *Message { - return &Message{parentCtx: parentCtx} +func NewMessage(reqID string, container Container, header Header) *Message { + return &Message{RequestID: reqID, Container: container, Header: header} } // Validate checks the integrity and validity of a Message object. From 28986707811621cb7a6c6319c0a6c35921cf4c04 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Wed, 11 Dec 2024 01:00:16 +0300 Subject: [PATCH 29/40] encode interface --- pkg/encode.go | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/pkg/encode.go b/pkg/encode.go index 2075100..2485a6f 100644 --- a/pkg/encode.go +++ b/pkg/encode.go @@ -59,3 +59,14 @@ func (Encoding) Decode(msg *Message) (*nats.Msg, error) { // using the corresponding fields from the custom Message. return &nats.Msg{Subject: msg.Subject, Data: buffer.Bytes(), Header: nats.Header(msg.Header)}, nil } + +// Encode defines an interface for encoding a NATS message into a custom Message. +// It abstracts the process of converting NATS-level message representations +// into application-specific Message structures to support interoperability. +type Encode interface { + // Encode is responsible for transforming a NATS message into a custom Message. + // The resulting custom Message contains data and metadata extracted from the NATS message, + // formatted for use within the application's messaging system. + // Returns the custom Message if successful or an error if the transformation fails. + Encode(msg *nats.Msg) (*Message, error) +} From 06d7756bce89e86bba0baa64bff899b726103c6a Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Wed, 11 Dec 2024 01:00:42 +0300 Subject: [PATCH 30/40] implement encode interface --- pkg/encode.go | 41 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/pkg/encode.go b/pkg/encode.go index 2485a6f..0d4e795 100644 --- a/pkg/encode.go +++ b/pkg/encode.go @@ -70,3 +70,44 @@ type Encode interface { // Returns the custom Message if successful or an error if the transformation fails. Encode(msg *nats.Msg) (*Message, error) } + +// Encode implements the Encode interface to convert a NATS message into a custom Message structure. +// This function processes the raw data and metadata from the NATS message and transforms it into +// a structured custom Message that is compatible with the application's requirements. +func (Encoding) Encode(msg *nats.Msg) (*Message, error) { + // Checks if the provided Message object is nil. + // If it is nil, an error is returned immediately, indicating the message is empty. + if msg == nil { + return nil, errors.New("message is empty") + } + + // Initializes a new buffer that will temporarily hold the JSON-encoded data. + // This buffer acts as the target for the encoding process. + buffer := new(bytes.Buffer) + + // Writes the raw data from the NATS message into the buffer. + // This step ensures the data is available for the JSON decoder. + // If writing to the buffer fails, the function returns the encountered error. + if _, err := buffer.Write(msg.Data); err != nil { + return nil, err + } + + // Initializes a JSON decoder to parse the data within the buffer. + // The decoder converts the JSON-formatted data into a Go structure. + decoder := json.NewDecoder(buffer) + + // Declares a variable to hold the result of decoding the JSON data. + // This variable will store the custom Message constructed from the decoded input. + var message Message + // Decodes the JSON data from the buffer into the custom Message structure. + // If the decoding process encounters an error, the function immediately returns it, + // signaling that the input data is not valid JSON or does not match the expected format. + if err := decoder.Decode(&message); err != nil { + return nil, err + } + + // Constructs a new custom Message using the decoded data and the header from the NATS message. + // The NewMessage function ensures that the Message is initialized with all required fields, + // including RequestID, Container, and Header. + return NewMessage(message.RequestID, message.Container, Header(msg.Header)), nil +} From 55322cb7caccc754288ed4276691c5f3aa18afb8 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Wed, 11 Dec 2024 01:02:40 +0300 Subject: [PATCH 31/40] add description --- pkg/message.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/pkg/message.go b/pkg/message.go index 02ce975..c38afab 100644 --- a/pkg/message.go +++ b/pkg/message.go @@ -62,6 +62,9 @@ type Message struct { parentCtx context.Context } +// NewMessage is a constructor function for creating a new instance of the Message structure. +// It initializes a Message object with the provided request ID, container, and header. +// This function simplifies the creation of Message instances by encapsulating the initialization logic. func NewMessage(reqID string, container Container, header Header) *Message { return &Message{RequestID: reqID, Container: container, Header: header} } From 73c783c820769303ea6e08499b5150c3b9d34497 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Wed, 11 Dec 2024 13:51:46 +0300 Subject: [PATCH 32/40] decode test case --- pkg/encode_test.go | 39 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/pkg/encode_test.go b/pkg/encode_test.go index 68f4d82..ad5bc51 100644 --- a/pkg/encode_test.go +++ b/pkg/encode_test.go @@ -4,6 +4,7 @@ import ( "bytes" "fmt" "testing" + "time" "github.com/nats-io/nats.go" "github.com/segmentio/encoding/json" @@ -92,3 +93,41 @@ func TestEncodingDecode(t *testing.T) { }) } } + +func TestEncode(t *testing.T) { + // Declare a variable of type Encoding. This variable will hold the instance of the encoding/decoding functionality + // that will be tested in the test cases. + var encoder Encoding + // Assert that the encoder variable is not nil. + // This check ensures that the Encoding instance has been properly initialized or is available for use. + // A nil value here would indicate a critical setup issue, rendering the test invalid. + assert.NotNil(t, encoder, "Expected encoder to be non-nil before running tests") + + // ValidNATSMessage tests the ability of the encoder to accurately decode and re-encode + // a custom Message to and from a NATS message. This test ensures the encoder handles + // valid data without errors and produces consistent results through the round-trip conversion. + // It validates the functionality of the Encode and Decode methods under normal conditions. + t.Run("ValidNATSMessage", func(t *testing.T) { + // Define a custom Message with valid test data for encoding and decoding. + // This Message includes a subject, request ID, container data, and a current timestamp. + message := &Message{Subject: "test.subject", Header: nil, RequestID: "6557162e-7a05-4840-a350-12a6f67e2b3b", Container: Container(`{"id": 1,"name": "Tammi Watson"}`), RequestTime: time.Now()} + + // Attempt to decode the custom Message into a NATS message. + // The Decode method transforms the application-level message into NATS-compatible format. + natsMsg, err := encoder.Decode(message) + // Assert that no error occurred during the decoding process. + // This verifies that valid messages are handled without issues during decoding. + assert.NoError(t, err, "Expected no error during decode operation") + // Assert that the resulting NATS message is not nil, indicating successful decoding. + assert.NotNil(t, natsMsg, "Expected a non-nil NATS message after decoding") + + // Attempt to encode the NATS message back into the custom Message format. + // This ensures the round-trip transformation is consistent and correct. + res, err := encoder.Encode(natsMsg) + // Assert that no error occurred during the encoding process. + // This confirms that valid NATS messages are handled correctly during encoding. + assert.NoError(t, err, "Expected no error during encode operation") + // Assert that the re-encoded Message is not nil, verifying successful encoding. + assert.NotNil(t, res, "Expected a non-nil custom message after encoding") + }) +} From 7a3efd0050c7c540fab8a1caef52130f589e8146 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Wed, 11 Dec 2024 13:54:27 +0300 Subject: [PATCH 33/40] empty message test case --- pkg/encode_test.go | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/pkg/encode_test.go b/pkg/encode_test.go index ad5bc51..c597c48 100644 --- a/pkg/encode_test.go +++ b/pkg/encode_test.go @@ -130,4 +130,18 @@ func TestEncode(t *testing.T) { // Assert that the re-encoded Message is not nil, verifying successful encoding. assert.NotNil(t, res, "Expected a non-nil custom message after encoding") }) + + // EmptyMessage tests the behavior of the Encode method when it is called with a nil message. + // It verifies that the method appropriately handles invalid input by returning an error. + // This test ensures that the Encode method is robust against edge cases and fails gracefully when given invalid data. + t.Run("EmptyMessage", func(t *testing.T) { + // Attempt to encode a nil message. + // This simulates a scenario where the Encode method is called without a valid Message object. + // The expected behavior is for the method to return an error, indicating the invalid input. + _, err := encoder.Encode(nil) + + // Assert that an error is returned during the encoding process when given a nil message. + // This ensures that the method validates input correctly and avoids processing invalid data. + assert.Error(t, err, "Expected an error when encoding a nil message") + }) } From 1fd002a801c36ef9b2760ffb1ab107aad42c4c60 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Wed, 11 Dec 2024 13:56:24 +0300 Subject: [PATCH 34/40] invalid json in nats message test --- pkg/encode_test.go | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/pkg/encode_test.go b/pkg/encode_test.go index c597c48..8f0cde4 100644 --- a/pkg/encode_test.go +++ b/pkg/encode_test.go @@ -144,4 +144,23 @@ func TestEncode(t *testing.T) { // This ensures that the method validates input correctly and avoids processing invalid data. assert.Error(t, err, "Expected an error when encoding a nil message") }) + + // InvalidJSONInNATSMessage tests the behavior of the Encode method when the input NATS message contains invalid JSON data. + // It verifies that the method correctly identifies the invalid JSON and returns an appropriate error. + // This test ensures the robustness of the Encode method in handling improperly formatted input data. + t.Run("InvalidJSONInNATSMessage", func(t *testing.T) { + // Create a NATS message containing invalid JSON in the Data field. + // This simulates a scenario where the message being processed has malformed JSON data, + // which the Encode method should be able to detect and reject. + invalidNATSMsg := &nats.Msg{Data: []byte(`invalid json`)} + + // Attempt to encode the NATS message containing invalid JSON. + // The Encode method is expected to fail and return an error in this scenario. + _, err := encoder.Encode(invalidNATSMsg) + + // Assert that an error is returned during the encoding process. + // This ensures the method correctly validates the JSON format in the input message + // and fails gracefully when invalid data is provided. + assert.Error(t, err, "Expected an error when encoding a NATS message with invalid JSON data") + }) } From bc0a99be559b883d973f960ad72cba96c5cb77ab Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Thu, 12 Dec 2024 00:30:05 +0300 Subject: [PATCH 35/40] change method name --- pkg/encode.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/pkg/encode.go b/pkg/encode.go index 0d4e795..e36d2af 100644 --- a/pkg/encode.go +++ b/pkg/encode.go @@ -60,18 +60,18 @@ func (Encoding) Decode(msg *Message) (*nats.Msg, error) { return &nats.Msg{Subject: msg.Subject, Data: buffer.Bytes(), Header: nats.Header(msg.Header)}, nil } -// Encode defines an interface for encoding a NATS message into a custom Message. +// Encoder defines an interface for encoding a NATS message into a custom Message. // It abstracts the process of converting NATS-level message representations // into application-specific Message structures to support interoperability. -type Encode interface { - // Encode is responsible for transforming a NATS message into a custom Message. +type Encoder interface { + // Encoder is responsible for transforming a NATS message into a custom Message. // The resulting custom Message contains data and metadata extracted from the NATS message, // formatted for use within the application's messaging system. // Returns the custom Message if successful or an error if the transformation fails. Encode(msg *nats.Msg) (*Message, error) } -// Encode implements the Encode interface to convert a NATS message into a custom Message structure. +// Encode implements the Encoder interface to convert a NATS message into a custom Message structure. // This function processes the raw data and metadata from the NATS message and transforms it into // a structured custom Message that is compatible with the application's requirements. func (Encoding) Encode(msg *nats.Msg) (*Message, error) { From fd42535bf4187f82e54df2d89325e7bdc1438545 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Thu, 12 Dec 2024 00:30:13 +0300 Subject: [PATCH 36/40] fix test --- pkg/encode_test.go | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/pkg/encode_test.go b/pkg/encode_test.go index 8f0cde4..5a5dddd 100644 --- a/pkg/encode_test.go +++ b/pkg/encode_test.go @@ -78,7 +78,7 @@ func TestEncodingDecode(t *testing.T) { // message into a JSON format for comparison against the 'Data' field of the result. var expectedData bytes.Buffer - // Encode the input message into the 'expectedData' buffer. This step ensures that the expected JSON data + // Encoder the input message into the 'expectedData' buffer. This step ensures that the expected JSON data // is correctly generated before comparing it with the actual 'Data' field in the result. err = json.NewEncoder(&expectedData).Encode(tt.input) @@ -106,7 +106,7 @@ func TestEncode(t *testing.T) { // ValidNATSMessage tests the ability of the encoder to accurately decode and re-encode // a custom Message to and from a NATS message. This test ensures the encoder handles // valid data without errors and produces consistent results through the round-trip conversion. - // It validates the functionality of the Encode and Decode methods under normal conditions. + // It validates the functionality of the Encoder and Decode methods under normal conditions. t.Run("ValidNATSMessage", func(t *testing.T) { // Define a custom Message with valid test data for encoding and decoding. // This Message includes a subject, request ID, container data, and a current timestamp. @@ -131,12 +131,12 @@ func TestEncode(t *testing.T) { assert.NotNil(t, res, "Expected a non-nil custom message after encoding") }) - // EmptyMessage tests the behavior of the Encode method when it is called with a nil message. + // EmptyMessage tests the behavior of the Encoder method when it is called with a nil message. // It verifies that the method appropriately handles invalid input by returning an error. - // This test ensures that the Encode method is robust against edge cases and fails gracefully when given invalid data. + // This test ensures that the Encoder method is robust against edge cases and fails gracefully when given invalid data. t.Run("EmptyMessage", func(t *testing.T) { // Attempt to encode a nil message. - // This simulates a scenario where the Encode method is called without a valid Message object. + // This simulates a scenario where the Encoder method is called without a valid Message object. // The expected behavior is for the method to return an error, indicating the invalid input. _, err := encoder.Encode(nil) @@ -145,17 +145,17 @@ func TestEncode(t *testing.T) { assert.Error(t, err, "Expected an error when encoding a nil message") }) - // InvalidJSONInNATSMessage tests the behavior of the Encode method when the input NATS message contains invalid JSON data. + // InvalidJSONInNATSMessage tests the behavior of the Encoder method when the input NATS message contains invalid JSON data. // It verifies that the method correctly identifies the invalid JSON and returns an appropriate error. - // This test ensures the robustness of the Encode method in handling improperly formatted input data. + // This test ensures the robustness of the Encoder method in handling improperly formatted input data. t.Run("InvalidJSONInNATSMessage", func(t *testing.T) { // Create a NATS message containing invalid JSON in the Data field. // This simulates a scenario where the message being processed has malformed JSON data, - // which the Encode method should be able to detect and reject. + // which the Encoder method should be able to detect and reject. invalidNATSMsg := &nats.Msg{Data: []byte(`invalid json`)} // Attempt to encode the NATS message containing invalid JSON. - // The Encode method is expected to fail and return an error in this scenario. + // The Encoder method is expected to fail and return an error in this scenario. _, err := encoder.Encode(invalidNATSMsg) // Assert that an error is returned during the encoding process. From e851731b7694ada91d1108c18b51f1f036c4d525 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Thu, 12 Dec 2024 13:11:19 +0300 Subject: [PATCH 37/40] refactoring message constructor --- pkg/message.go | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/pkg/message.go b/pkg/message.go index c38afab..20e68b8 100644 --- a/pkg/message.go +++ b/pkg/message.go @@ -55,18 +55,15 @@ type Message struct { // once is used to ensure certain operations are performed only once. // It uses sync.Once to guarantee that specific actions, such as closing // the channel and unsubscribing, are executed only a single time. - once sync.Once + once *sync.Once // parentCtx is the context associated with the message's parent operation. // It provides a way to propagate cancellation, timeouts, or deadlines across operations. parentCtx context.Context } -// NewMessage is a constructor function for creating a new instance of the Message structure. -// It initializes a Message object with the provided request ID, container, and header. -// This function simplifies the creation of Message instances by encapsulating the initialization logic. -func NewMessage(reqID string, container Container, header Header) *Message { - return &Message{RequestID: reqID, Container: container, Header: header} +func NewMessage(parentCtx context.Context) *Message { + return &Message{parentCtx: parentCtx, once: new(sync.Once)} } // Validate checks the integrity and validity of a Message object. From ad048f9f4c3e5822a680d44f10127dadeb84bb8a Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Thu, 12 Dec 2024 13:12:34 +0300 Subject: [PATCH 38/40] fix encoder (constructor isn't use) --- pkg/encode.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/encode.go b/pkg/encode.go index e36d2af..01e4487 100644 --- a/pkg/encode.go +++ b/pkg/encode.go @@ -109,5 +109,5 @@ func (Encoding) Encode(msg *nats.Msg) (*Message, error) { // Constructs a new custom Message using the decoded data and the header from the NATS message. // The NewMessage function ensures that the Message is initialized with all required fields, // including RequestID, Container, and Header. - return NewMessage(message.RequestID, message.Container, Header(msg.Header)), nil + return &message, nil } From 43aad7465e5955320648986934d9c78874db6b3c Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Thu, 12 Dec 2024 15:55:13 +0300 Subject: [PATCH 39/40] update encoder test --- pkg/encode_test.go | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/pkg/encode_test.go b/pkg/encode_test.go index 5a5dddd..8948a89 100644 --- a/pkg/encode_test.go +++ b/pkg/encode_test.go @@ -129,6 +129,18 @@ func TestEncode(t *testing.T) { assert.NoError(t, err, "Expected no error during encode operation") // Assert that the re-encoded Message is not nil, verifying successful encoding. assert.NotNil(t, res, "Expected a non-nil custom message after encoding") + // Verify that the request ID of the re-encoded Message matches the original custom Message. + // This confirms that the request ID field is preserved correctly during the round-trip transformation. + assert.Equal(t, message.RequestID, res.RequestID, "Expected the RequestID to match the original message") + // Verify that the subject of the re-encoded Message matches the original custom Message. + // This ensures that the subject field remains consistent after the transformation. + assert.Equal(t, message.Subject, res.Subject, "Expected the Subject to match the original message") + // Verify that the container of the re-encoded Message matches the original custom Message. + // This validates that the container data is preserved during encoding and decoding. + assert.Equal(t, message.Container, res.Container, "Expected the Container to match the original message") + // Verify that the header of the re-encoded Message matches the original custom Message. + // This confirms that the header field is not altered during the encoding and decoding processes. + assert.Equal(t, message.Header, res.Header, "Expected the Header to match the original message") }) // EmptyMessage tests the behavior of the Encoder method when it is called with a nil message. From 857effd98f81d7a30c05001ec01ff638d82e4335 Mon Sep 17 00:00:00 2001 From: syntaxErrorLineNil Date: Thu, 12 Dec 2024 17:48:41 +0300 Subject: [PATCH 40/40] add nats message in message object --- pkg/encode.go | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/pkg/encode.go b/pkg/encode.go index 01e4487..ebce2e7 100644 --- a/pkg/encode.go +++ b/pkg/encode.go @@ -64,7 +64,7 @@ func (Encoding) Decode(msg *Message) (*nats.Msg, error) { // It abstracts the process of converting NATS-level message representations // into application-specific Message structures to support interoperability. type Encoder interface { - // Encoder is responsible for transforming a NATS message into a custom Message. + // Encode is responsible for transforming a NATS message into a custom Message. // The resulting custom Message contains data and metadata extracted from the NATS message, // formatted for use within the application's messaging system. // Returns the custom Message if successful or an error if the transformation fails. @@ -106,8 +106,14 @@ func (Encoding) Encode(msg *nats.Msg) (*Message, error) { return nil, err } - // Constructs a new custom Message using the decoded data and the header from the NATS message. - // The NewMessage function ensures that the Message is initialized with all required fields, - // including RequestID, Container, and Header. + // Associates the original NATS message with the newly decoded custom Message. + // This linkage allows the application to reference the original NATS message + // when working with the custom Message structure. + message.message = msg + + // Constructs and returns a pointer to the custom Message populated with the decoded data. + // The NewMessage function ensures all fields, such as RequestID, Container, and Header, + // are initialized correctly. This completes the transformation of the NATS message into + // the custom Message format required by the application. return &message, nil }