Merge pull request #5 from wingify/feat/ordering-key

Added support for ordering key while publishing message

CHANGELOG.md

@@ -4,6 +4,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [1.5] - 2023-11-13
+### Added
+- Added support for ordering key while publishing message
+
 ## [1.4] - 2021-04-17
 ### Fixed
 - Added exception handling in push_batch method for releasing lock

README.md

@@ -103,7 +103,7 @@ Note that at least [ngx_lua 0.9.3](https://github.com/openresty/lua-nginx-module
                     local ok, send_err = producer:send("Some Random Text", {
                         attr1 = "Test1",
                         attr2 = "Test2"
-                    })
+                    }, "optional_ordering_key")
 
                     -- Also check if there is any error while sending message
                     if send_err ~= nil then
@@ -167,9 +167,9 @@ To load this module, just do this
 `syntax: local p, err = producer:new(pubsub_config)`
 
 #### send
-`syntax: p:send(message, attributes)`
+`syntax: p:send(message, attributes[, ordering_key])`
 
-* Requires a message of type string and attributes of type table
+* Requires a message of type string, attributes of type table and an optional ordering_key of type string
 
 [Back to TOC](#table-of-contents)
 

lib/resty/pubsub/producer.lua

@@ -149,7 +149,7 @@ _timer_flush = function (premature, self, time)
     end
 end
 
-function _M.send(self, message, attributes)
+function _M.send(self, message, attributes, ordering_key)
 
     if type(message) ~= "string" then
         return false, "Data expected in string, got " .. type(message)
@@ -165,6 +165,15 @@ function _M.send(self, message, attributes)
         attributes = attributes
     }
 
+
+    if ordering_key ~= nil then
+        if type(ordering_key) == "string" then
+            body_message['ordering_key'] = ordering_key
+        else
+            ngx.log(ngx.DEBUG, "Ordering key must be String hence dropping it ", ordering_key)
+        end
+    end
+
     if self.ring_buffer == nil then
         return false, "Buffer not initialized Properly"
     end