Prefetch count configurability (#8)

* Add prefetchCount configurability.
* Make prefetchCount have a default value of 10.

Co-authored-by: Rasmus Larsson <[email protected]>
Co-authored-by: Emil Klasson <[email protected]>
Co-authored-by: Kristian Lundström <[email protected]>

Author: 4 people

Makefile

@@ -1,20 +1,14 @@
 PROJECT_NAME=Insurello.RabbitMqClient
-TEST_PROJECT_NAME=$(PROJECT_NAME).Tests
 PROJECT_DIR=src/$(PROJECT_NAME)
-TEST_DIR=src/$(TEST_PROJECT_NAME)
 MAIN_PROJ=$(MAIN_DIR)/$(PROJECT_NAME).fsproj
-TEST_PROJ=$(TEST_DIR)/$(TEST_PROJECT_NAME).fsproj
 
 
-.PHONY : all build test
+.PHONY : all build
 
-all: build test
+all: build
 
-test:
-	dotnet test $(TEST_PROJ)
-
-test-watch:
-	dotnet watch --project $(TEST_DIR) run
+build:
+	dotnet build
 
 clean:
-	rm -r $(PROJECT_DIR)/bin $(PROJECT_DIR)/obj $(TEST_DIR)/bin $(TEST_DIR)/obj
+	rm -r $(PROJECT_DIR)/bin $(PROJECT_DIR)/obj 

src/Insurello.RabbitMqClient/MqClient.fs

@@ -57,8 +57,11 @@ module MqClient =
           Headers: Map<string, string>
           Content: Content }
 
-    type ChannelConfig =
-        { withConfirmSelect: bool }
+    type private ChannelConfig =
+        { withConfirmSelect: bool
+          prefetchCount: uint16 }
+          
+    type PrefetchCount = PrefetchCount of uint16
 
     let private contentTypeStringFromContent: Content -> string =
         function
@@ -171,6 +174,7 @@ module MqClient =
         fun config exCallback connection ->
             try
                 let model = connection.CreateModel()
+                model.BasicQos(uint32 0, config.prefetchCount,false)
 
                 if config.withConfirmSelect then model.ConfirmSelect()
 
@@ -306,18 +310,23 @@ module MqClient =
 
     let messageId: ReceivedMessage -> string = fun (Message(event, _)) -> event.BasicProperties.MessageId
 
-    let init: LogError -> string -> System.Uri -> (Model -> Topology) -> Result<Model, string> =
-        fun logError nameOfClient uri getTopology ->
+    let init: LogError -> string -> System.Uri -> Option<PrefetchCount> -> (Model -> Topology) -> Result<Model, string> =
+        fun logError nameOfClient uri prefetchCountOption getTopology ->
+            let prefetchCount = 
+                prefetchCountOption 
+                |> Option.map (fun (PrefetchCount prefetchCount) -> prefetchCount)
+                |> Option.defaultValue (uint16 10)
+                                
             connect nameOfClient uri
             |> Result.bind (fun connection ->
                 let exCallback =
                     (fun ex context connection ->
                         logError (ex, "Unhandled exception on channel in context {$c}", context)
                         closeConnectionAsync (System.TimeSpan.FromSeconds 3.0) connection)
 
-                createChannel { withConfirmSelect = true } exCallback connection
+                createChannel { withConfirmSelect = true; prefetchCount = prefetchCount } exCallback connection
                 |> Result.bind (fun channel ->
-                    createChannel { withConfirmSelect = false } exCallback connection
+                    createChannel { withConfirmSelect = false; prefetchCount = prefetchCount } exCallback connection
                     |> Result.map (fun rpcChannel ->
                         Model
                             { channelConsumer = AsyncEventingBasicConsumer channel