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") -) diff --git a/go.mod b/go.mod index 1918b42..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,6 +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/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 aabedab..528831e 100644 --- a/go.sum +++ b/go.sum @@ -16,6 +16,10 @@ 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= github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= golang.org/x/crypto v0.18.0 h1:PGVlW0xEltQnzFZ55hkuX5+KLyrMYhHld1YHO4AKcdc= diff --git a/pkg/encode.go b/pkg/encode.go new file mode 100644 index 0000000..ebce2e7 --- /dev/null +++ b/pkg/encode.go @@ -0,0 +1,119 @@ +package pkg + +import ( + "bytes" + "errors" + + "github.com/nats-io/nats.go" + "github.com/segmentio/encoding/json" +) + +// 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) { + // 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) + // 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 +} + +// 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 Encoder 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) +} + +// 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) { + // 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 + } + + // 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 +} diff --git a/pkg/encode_test.go b/pkg/encode_test.go new file mode 100644 index 0000000..8948a89 --- /dev/null +++ b/pkg/encode_test.go @@ -0,0 +1,178 @@ +package pkg + +import ( + "bytes" + "fmt" + "testing" + "time" + + "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 + + // 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) + + // 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)) + } + }) + } +} + +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 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. + 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") + // 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. + // It verifies that the method appropriately handles invalid input by returning an error. + // 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 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) + + // 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") + }) + + // 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 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 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 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. + // 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") + }) +} diff --git a/pkg/message.go b/pkg/message.go new file mode 100644 index 0000000..20e68b8 --- /dev/null +++ b/pkg/message.go @@ -0,0 +1,220 @@ +package pkg + +import ( + "context" + "sync" + "time" + + pubsub "github.com/SyntaxErrorLineNULL/nats-pubsub-go" + "github.com/nats-io/nats.go" +) + +// 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 { + // 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"` + + // Container holds the payload of the message as a byte slice. + // It represents the data being transmitted or processed in the communication. + 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 + // 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 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(parentCtx context.Context) *Message { + return &Message{parentCtx: parentCtx, once: new(sync.Once)} +} + +// 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. +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 +} + +// 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. +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 +} + +// 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) +} + +// 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() + } +} + +// 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) +} + +// 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) +} diff --git a/pkg/subscriber.go b/pkg/subscriber.go index 10e6d3e..9a623db 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,24 @@ func NewSubscriber(conn *nats.Conn) *Subscriber { return &Subscriber{conn: conn} } +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 + } + + // 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 +} + // 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. 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.