在SubscriptionClient上调用RegisterMessageHandler时出现超时异常

Question

我试图在运行在Android中的SubscriptionClient实例上调用RegisterMessageHandler方法。

在将消息发布到特定主题（通过自动测试）20秒左右之后，我在我的Android应用程序中收到了几个超时异常，该应用程序实际上包含了SubscriptionClient实例的RegisterMessageHandler。

输出窗口

这里有个错误：

**System.TimeoutException：**“Timeout”

member x.Enable () =

    async {

        if not (isEnabled) then

            subscriptionClient <- new SubscriptionClient(connectionString, "Topic.courier-requested", "Subscription.all-messages")
            subscriptionClient.OperationTimeout <- TimeSpan.FromMinutes(1.0)

            let! rulesFound     = subscriptionClient.GetRulesAsync() |> Async.AwaitTask
            let  hasDefaultRule = rulesFound.Any(fun r -> r.Name = RuleDescription.DefaultRuleName)

            if hasDefaultRule then
                do! subscriptionClient.RemoveRuleAsync(RuleDescription.DefaultRuleName) |> Async.AwaitTask

            else
                let msgOptions = MessageHandlerOptions(fun args -> exceptionReceivedHandler(args))
                msgOptions.AutoComplete         <- false
                msgOptions.MaxAutoRenewDuration <- TimeSpan.FromMinutes(1.0)
                msgOptions.MaxConcurrentCalls   <- 1

                subscriptionClient.RegisterMessageHandler(processMessageAsync, msgOptions)

    } |> Async.StartAsTask

let processMessageAsync (message:Message) (_:CancellationToken) = 

    let json    = Encoding.UTF8.GetString(message.Body)
    ...

    subscriptionClient.CompleteAsync(message.SystemProperties.LockToken) |> Async.AwaitTask |> Async.RunSynchronously
    Task.CompletedTask

Android测试-失败

下面的测试发布了一条消息，该消息在大约20秒后导致我的Android应用出现超时异常：

[<Fact>]
let ``Publish message courier-requested message to servicebus topic``() =

    // Setup
    let connectionstring = ConfigurationManager.ConnectionStrings.["servicebus_testEnv"].ConnectionString

    // Test
    async {

        let client    = TopicClient(connectionstring, "Topic.courier-requested")
        let data      = "test_data"
        let message   = Message(Encoding.UTF8.GetBytes(data))
        let courierId = "b965f552-31a4-4644-a9c6-d86dd45314c4"
        message.Label <- sprintf "courier-id(%s)" courierId

        do! client.SendAsync(message) |> Async.AwaitTask
        do! client.CloseAsync()       |> Async.AwaitTask
    }

控制台测试-通过

[<EntryPoint>]
let main argv =

    printfn "Welcome to Subscription.Console"

    async {

        let subscriber = Subscriber(connectionString)
        do! subscriber.Listen()

    } |> Async.RunSynchronously


    Console.ReadKey() |> ignore
    0 // return an integer exit code

type Subscriber(connectionString:string) =

    let mutable subscriptionClient : SubscriptionClient = null

    let exceptionReceivedHandler (args:ExceptionReceivedEventArgs) =
        printfn "Got an exception: %A" args.Exception
        Task.CompletedTask

    let processMessageAsync (message:Message) (_:CancellationToken) = 

        let json = Encoding.UTF8.GetString(message.Body)

        subscriptionClient.CompleteAsync(message.SystemProperties.LockToken) |> Async.AwaitTask |> Async.RunSynchronously
        Task.CompletedTask

    member x.Listen() =

        async {

            subscriptionClient <- new SubscriptionClient(connectionString, "Topic.courier-requested", "Subscription.all-messages")
            subscriptionClient.OperationTimeout <- TimeSpan.FromMinutes(1.0)

            let! rulesFound     = subscriptionClient.GetRulesAsync() |> Async.AwaitTask
            let  hasDefaultRule = rulesFound.Any(fun r -> r.Name = RuleDescription.DefaultRuleName)

            if hasDefaultRule then
                do! subscriptionClient.RemoveRuleAsync(RuleDescription.DefaultRuleName) |> Async.AwaitTask

            else

                let msgOptions = MessageHandlerOptions(fun args -> exceptionReceivedHandler(args))
                msgOptions.AutoComplete         <- false
                msgOptions.MaxAutoRenewDuration <- TimeSpan.FromMinutes(1.0)
                msgOptions.MaxConcurrentCalls   <- 1

                subscriptionClient.RegisterMessageHandler(processMessageAsync, msgOptions)
        }

我将以下内容应用于我的连接字符串：

TransportType=AmqpWebSockets;

我参考了以下链接：

https://github.com/azure/azure-service-bus-dotnet/issues/529

Answer 1

Android模拟器没有连接到Internet。