时间戳

首页 > 解决方案 > 可观察的不包含“哪里”的定义

c# - 可观察的不包含“哪里”的定义

问题描述

我正在尝试使用这个在我的 Xamarin Android 应用程序中实现 MQTT 。但是我收到错误

' IObservable<MqttApplicationMessage>' 不包含 'Where' 的定义，并且找不到接受第一个类型为 ' IObservable<MqttApplicationMessage>' 的参数的可访问扩展方法 'Where'（您是否缺少 using 指令或程序集引用？）。

尝试删除Where只是为了看看它是否会起作用，而是得到

无法将 lambda 表达式转换为类型“ IObserver<MqttApplicationMessage>”，因为它不是委托类型。

尝试使用 Ably.IO 和 PubNub 并能够实现它的发布/订阅功能，但由于它的消息速率限制，我正在尝试这个并希望它没有消息速率限制。（例如：Ably 30-35msg/s 限制）

using System.Collections.Generic;
using System.Text;
using Android.App;
using Android.OS;
using Android.Widget;
using OxyPlot;
using OxyPlot.Series;
using OxyPlot.Xamarin.Android;
using OxyPlot.Axes;
using PubnubApi;
using System.Linq;
using Java.Lang;
using IO.Ably;
using IO.Ably.Realtime;
using System.Drawing;
using System;
using System.Runtime.Serialization.Formatters.Binary;
using System.IO;
using System.ComponentModel;
using System.Net.Mqtt;
using System.Reactive.Linq;
using System.Threading.Tasks;

protected override async void OnCreate(Bundle savedInstanceState)
{ 
    base.OnCreate(savedInstanceState);

    var configuration = new MqttConfiguration
    {
        //  BufferSize = 128 * 1024,
        Port = 1883,
        KeepAliveSecs = 10,
        WaitTimeoutSecs = 2,
        MaximumQualityOfService = MqttQualityOfService.AtMostOnce,
        AllowWildcardsInTopicFilters = true
    };
    var client = await MqttClient.CreateAsync("test.mosquitto.org", configuration);
    var sessionState = await client.ConnectAsync
        (new MqttClientCredentials(clientId: "foo"), cleanSession: true);
    await client.SubscribeAsync("foo/bar/sampletopic", 
        MqttQualityOfService.AtMostOnce); //QoS0

    client.MessageStream.Where(msg => msg.Topic == "foo/bar/sampletopic").
      Subscribe(msg =>RcvdMsgtext.text=Encoding.ASCII.GetString(msg.payload) );

    pubbtn.Click += async delegate
    {

        var message1 = new MqttApplicationMessage(
            "foo/bar/sampletopic",
            Encoding.UTF8.GetBytes("Foo Message 1"));
        await client.PublishAsync(message1, MqttQualityOfService.AtMostOnce);
    }

//note: Removed some parts but basically all i did was placed  it on OnCreate, changed it
//to async. In using Ably or PubNub "async" is not needed in OnCreate

期望在单击发布按钮时显示消息/文本更改。

编辑：能够通过检查我的引用绕过这个错误，看到 System.Reactive 带有黄色图标，双击它并且 IObservable 错误消失了，但是产生了需要我添加 System.Threading.Tasks.Extensions 的错误，我在版本 4.0.0（阅读有关使用此版本进行构建的内容）。

编辑：在代码中添加了使用..

现在只要代码运行通过这一行：

var client = await MqttClient.CreateAsync("test.mosquitto.org",       configuration);
       var sessionState = await client.ConnectAsync
       (new MqttClientCredentials(clientId: "foo"), cleanSession: true);

我收到一个错误

System.TypeLoadException：无法从 typeref 解析带有令牌 01000023 的类型（程序集“System.Core，版本=2.0.5.0，文化=中性，PublicKeyToken=7cec85d7bea7798e”中的预期类“System.Reactive.Concurrency.IScheduler”）

标签： c#xamarin.androidmqtt

解决方案

确保 System.Reactive.Linq 在 Using 部分代码中，以及在引用中。对于 VS 2017 可能使用较低版本，因为这还需要您添加 System.Threading.Tasks.Extensions

还使用 M2MQTT 作为替代方案。

推荐阅读