fsprojects / fsharp.cloudagent Goto Github PK

Currently you must post messages to pools one at a time. There should be the option to provide a set of messages that can be posted in a batch.

Subscriptions are the bomb

In contrast to queues, in which each message is processed by a single consumer, topics and subscriptions provide a one-to-many form of communication, in a “publish/subscribe” pattern. Useful for scaling to very large numbers of recipients, each published message is made available to each subscription registered with the topic. Messages are sent to a topic and delivered to one or more associated subscriptions, depending on filter rules that can be set on a per-subscription basis. The subscriptions can use additional filters to restrict the messages that they want to receive. Messages are sent to a topic in the same way they are sent to a queue, but messages are not received from the topic directly. Instead, they are received from subscriptions. A topic subscription resembles a virtual queue that receives copies of the messages that are sent to the topic. Messages are received from a subscription identically to the way they are received from a queue.

By way of comparison, the message-sending functionality of a queue maps directly to a topic and its message-receiving functionality maps to a subscription

If a Service Bus session "expires" through no activity, the first time you send a new message to that session, it can take up to a minute for it to be handled.

@isaacabraham - the usual expected issue from me :)

Could we consider renaming this to FSharp.Azure.CloudAgent? It seems the logical and consistent name for the framework?

I'm able to add messages to the queue but when I try to receive them nothing is received.

I've used the Azure Service Bus in linqpad with C# sample code so I know the queue works, but I'm probably missing something on the F# implementation with CloudAgent.

I tried serializing the object before adding it to the queue so that the content would just be a string that I could deserialize on receipt, but still had the same issue.

If I set a breakpoint on the line inbox.Receive() it gets hit, but a breakpoint on the following line never does.

Thanks for the help.

I'm not sure if it'd be more helpful to put the code in this issue or point you to my project so here are both.
https://github.com/talbottmike/TalBot/tree/AzureServiceBus

The type I'm posting is in TalBot.Types...

type IncomingMessage = 
    {
        token : string;
        teamId : string;
        teamDomain : string;
        channelId : string;
        channelName : string;
        timestamp : decimal;
        userId : string;
        userName : string;
        text : string;
        triggerWord : string;
    }

I'm accessing the queue in TalBot.BotHelper around the beginning of the module.

let createResilientAgent agentId =
    MailboxProcessor.Start(fun inbox ->
        async {
            while true do
                let! message, replyChannel = inbox.Receive()
                printfn "%s is the channelName." message.channelName
                printfn "%s" (agentId.ToString())

                match message with
                | { channelName = "#bot-log" } -> 
                    printfn "success"
                    replyChannel Completed // all good, message was processed
                | { channelName = "snapple" } -> 
                    printfn "snapple failed"
                    replyChannel Failed // error occurred, try again
                | _ -> 
                    printfn "abandoned"
                    replyChannel Abandoned // give up with this message.
        })

let readFromServiceQueue () =
    let serviceBusReadConnection () = ServiceBusConnection serviceBusReadConnectionString
    let cloudReadConnection () = WorkerCloudConnection(serviceBusReadConnection (), Queue "queue")
    printfn "trying to read from the queue"
    let disposable = ConnectionFactory.StartListening(cloudReadConnection (), createResilientAgent >> Messaging.CloudAgentKind.ResilientCloudAgent)
    disposable.Dispose()    

On long running tasks it is necessary to extend visibility timeouts to prevent them being incorrectly redelivered.

To achieve this you need either the CloudQueueMessage in scope, or the MessageId or SequenceNumber and also LockToken in order to use raw REST API.

via .NET client you'd do this:
https://msdn.microsoft.com/en-us/library/azure/microsoft.windowsazure.storage.queue.cloudqueue.updatemessage.aspx

via REST you'd do this:
https://msdn.microsoft.com/en-us/library/azure/jj839746.aspx

Since the CloudAgent paradigm abstracts away some of these details, a long running agent will incorrectly have the same message delivered multiple times.

It could be that a "ReplyChannel Extend" type operation is a relevant solution.

There is already support under-the-hood for different serializers as well as controlling the number of agents in worker pools, but this is currently not surfaced to the caller.

"Service Bus for Windows Server is a set of installable components that provides the messaging capabilities of Windows Azure Service Bus on Windows. Service Bus for Windows Server enables you to build, test, and run loosely-coupled, message-driven applications in self-managed environments and on developer computers. Service Bus queues offer reliable message storage and retrieval with a choice of protocols and APIs. Building on the same foundation as queues, Service Bus topics provide rich publish/subscribe capabilities that allow multiple, concurrent subscribers to independently retrieve filtered or unfiltered views of the published message stream." http://msdn.microsoft.com/en-us/library/dn282144.aspx

NOTE: There may be minor incompatibilities between the DLL versions used by Windows and Asure - see: https://code.msdn.microsoft.com/windowsapps/service-bus-explorer-f2abca5a

Maybe it's possible to track some status of current (computations) operations (e.g. Completed | Failed | Waiting)

Hi @isaacabraham - since AzureStorageTypeProvider/ moved to http://github.com/fsprojects, would you like to move this there too?

cheers!
don

Due to the date of last commit I wonder if the project still is alive/usable.

Perhaps create a script or side project that can create a service bus, or perhaps a sample project that shows how to use CloudAgent in an end-to-end manner.