M2MQTT клиент отключается без исключения или сообщения об ошибке - PullRequest
Новая
       9

M2MQTT клиент отключается без исключения или сообщения об ошибке

0 голосов
/ 05 июня 2018

Я пытаюсь создать API, который использует различные темы.Для этого я пытаюсь многопоточность, чтобы потом можно было масштабировать все в несколько API, но это очень важно.

Я использую ASP.net Core 4.0, если это как-то связано с этим.Entity Framework.

Моя проблема заключается в том, что мое соединение с моим сервером Mosquitto было разорвано без исключения или чего-то подобного, через минуту или около того.Неважно, насколько велики сообщения или сколько их обменивают.Я понятия не имею, как я могу создать обратный вызов или что-то в этом роде, чтобы узнать, что происходит с моей связью.Кто-нибудь может помочь?

Я свяжу код, который я использую для установления соединения, и подпишусь на соединение ниже.Использование метода подписки или выполнение его вручную также ничего не меняет.Я в растерянности, здесь.

Заранее спасибо!

Main.cs: 

Task.Factory.StartNew(() => DataflowController.ResumeQueuesAsync());
BuildWebHost(args).Run();

DataflowController.cs: 

public static Boolean Subscribe(String topic)
    {

        Console.WriteLine("Hello from " + topic);

        MqttClient mqttClient = new MqttClient(brokerAddress);
        byte code = mqttClient.Connect(Guid.NewGuid().ToString());

        // Register to message received
        mqttClient.MqttMsgPublishReceived += client_recievedMessageAsync;

        string clientId = Guid.NewGuid().ToString();
        mqttClient.Connect(clientId);

        // Subscribe to topic
        mqttClient.Subscribe(new String[] { topic }, new byte[] { MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE });
        System.Console.ReadLine();

        return true;
    }

    public static async Task ResumeQueuesAsync()
    {
        var mongoClient = new MongoClient(connectionString);
        var db = mongoClient.GetDatabase(databaseName);

        var topics = db.GetCollection<BsonDocument>(topicCollection);
        var filter = new BsonDocument();
        List<BsonDocument> result = topics.Find(filter).ToList();
        var resultSize = result.Count;
        Task[] subscriptions = new Task[resultSize];

        MqttClient mqttClient = new MqttClient(brokerAddress);
        byte code = mqttClient.Connect(Guid.NewGuid().ToString());

        // Register to message received
        mqttClient.MqttMsgPublishReceived += client_recievedMessageAsync;

        string clientId = Guid.NewGuid().ToString();
        mqttClient.Connect(clientId);


        int counter = 0;
        foreach(var doc in result)
        {

            subscriptions[counter] = new Task(() =>
            {
                Console.WriteLine("Hello from " + doc["topic"].ToString());
                // Subscribe to topic
                mqttClient.Subscribe(new String[] { doc["topic"].ToString() }, new byte[] { MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE });
                System.Console.ReadLine();
            });
            counter++;
        }

        foreach(Task task in subscriptions)
        {
            task.Start();
        }
    }

    static async void client_recievedMessageAsync(object sender, MqttMsgPublishEventArgs e)
    {
        // Handle message received
        var message = System.Text.Encoding.Default.GetString(e.Message);
        var topic = e.Topic;

        var id = topic.Split("/")[2];

        BsonDocument doc = new BsonDocument {
            {"Plug ID", id },
            {"Consumption", message }
        };

        await Save(doc, "smartPDM_consumption");

        System.Console.WriteLine("Message received from " + topic + " : " + message);
    }

1 Ответ

0 голосов
/ 12 июля 2018

Эта строка была проблемой: 

byte code = mqttClient.Connect(Guid.NewGuid().ToString());

Удалил ее, и она просто заработала.

...