tracing and ids alignment

Author: et1975

src/DatatypeChannels.ASB/Consumer.fs

@@ -37,7 +37,9 @@ let mkNew (options:ServiceBusReceiverOptions)
         match ctxRef.Value with
         | Some ((receiver,_,_) as ctx) when not receiver.IsClosed -> Task.FromResult ctx
         | _ -> fun (name:Task<_>) -> backgroundTask {
+                    use activity = activitySource |> Diagnostics.startActivity "new" ActivityKind.Consumer
                     let! name = name
+                    activity.AddTag("binding", $"{name}/{options.SubQueue}") |> ignore
                     let startActivity _ = activitySource |> Diagnostics.startActivity name ActivityKind.Consumer
                     let receiver = withClient (fun client -> client.CreateReceiver(name, options))
                     ctxRef.Value <- Some (receiver, startActivity, ConcurrentDictionary())
@@ -71,9 +73,9 @@ let mkNew (options:ServiceBusReceiverOptions)
                         return msg |> Option.map (fun msg -> 
                             let msgCtx = MessageContext.From received activity receiver
                             if receiver.ReceiveMode = ServiceBusReceiveMode.PeekLock then
-                                if msgCtxs.TryAdd(received.LockToken, msgCtx) then startRenewal msgCtx |> ignore
+                                if msgCtxs.TryAdd(received.MessageId, msgCtx) then startRenewal msgCtx |> ignore
                                 else failwith "Unable to add the message"
-                            { Msg = msg; Id = received.LockToken })
+                            { Msg = msg; Id = received.MessageId })
                 }
 
             member __.Ack receivedId =

tests/DatatypeChannels.ASB.Tests/Tests.fs

@@ -153,13 +153,15 @@ let tests =
                 let! dlc = DeadLetter "renews-queue" |> channels.GetConsumer PlainText.ofReceived
                 let publisher = channels.GetPublisher (PlainText.toSend testId) topic
                 do! Task.Delay 5_000 // majic number - this is how long it takes the backend to start routing messages to this subscription!
-                let n = 20
+                let n = 200
                 for i in 1..n do
                     do! publisher |> Publisher.publish $"test-payload-{i}"
-                let xs = ResizeArray()
-                for i in 1..n do
-                    let! received = TimeSpan.FromSeconds 1. |> consumer.Get
-                    received |> Option.iter xs.Add 
+                let! xs = // also test parallel consumption
+                    seq {for _ in 1..2 -> seq { for _ in 1..n/2 -> TimeSpan.FromSeconds 1. |> consumer.Get}}
+                    |> Seq.map Task.WhenAll
+                    |> Task.WhenAll
+                    |> Task.map (Array.collect id)
+                    |> Task.map (Array.choose id)
                 xs |> Seq.distinct |> Seq.length =! n
                 do! Task.Delay (TimeSpan.FromMinutes 7.)
                 for received in xs do