Гyдвинволшебник (13.05.2025 16:21, просмотров: 43) ответил vesago на Коллеги, присоветуйте. Есть плата с выходом RS-485. Нужно приделать
по MQTT. Правильно я думаю, что верный путь - мастерить преобраз
RS-485 в изернет. И по изернету к брокеру подкидывать? Или есть
какие-то тулзы. может брокер, к которому можно напрямую по
последовательному интерфейсу подкинуть?Есть вот такая библиотека. Изращался с неким бриджем для работы 1С
с MQTT. Пользовал правда UDP протокол для связи. Типа запущал свою
приложуху сервером, который перенапралял UDP пакеты MQTT брокеру и
наоборот). Вместо UDP можно заюзать UART. А извращался потому как
это было еще во времена XP и совокупить библиотеку удалось только с
компилятором PellesC ;)
/*******************************************************************************
служебная строка при запуске:
mqservice.exe m24.cloudmqtt.com:13310 dazkzemv AWz0stwtWlNX qwert/bsz/90788555/inp
сообщения, приходящие в 1C GSevent:
15,0,qwert/bsz/90788555/inp,f5dfdfdfdfdfdfdf337833 - приняты данные подписки
15,1,connection lost - нет соединения с mqtt сервером
15,2,MQTT connected - произошло соедиенеие с mqtt сервером
15,3,Service stopped - стоп сервиса
команды из компа:
mqttqwert/bsz/90788555/inp,f5dfdfdfdfdfdfdf337833 - опубликовать в топик
mqttstop_mqttbr - завершение работы моста
ответы на команды:
mqtt0 - ок - команда выполнена
mqtt1 - error - соединение с mqtt сервером потеряно, команда НЕ выполнена
mqtt2 - error - неправильная команда, команда НЕ выполнена
//mqtt3 - error (в программе SCGiga1C) - таймаут/в сети не найден мост, команда НЕ выполнена
В SCGiga1C:
команда: 15 - Опубликовать строку в топик MQTT сервера
Giga.str_for_cmd = "holding/bsz/90788555/inp,data1,data2...." ; // строка c данными для публикации. До первой запятой - имя топика, остальное данные (не более 500 символов)
error = Giga.cmd_to_simcon(ip_addr,15,0); //
коды ошибок:
0 -OK
1- Потеряна связь с MQTT сервером
2 - Неверная строка
3 - Мост MQTT не найден в сети/не ответил в течении 300 мс
*******************************************************************************/
#include "stdio.h"
#include "stdlib.h"
#include "string.h"
#include "MQTTClient.h"
#include "windows.h"
#include <time.h>
#define IN_PORT 31203 // приемный порт mqttbr
#define PORT_1C 31202 // UDP порт для 1C event
#define VID 0x16c0
#define PID 0x05df
#define MSGBUFSIZE 1024
char in_msgbuf[MSGBUFSIZE];
char out_msgbuf[MSGBUFSIZE];
struct sockaddr_in addr_in;
struct sockaddr_in addr_out;
int fd_in;
int fd_out;
#define QOS 0
MQTTClient client;
MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
volatile int fl_connect=0;
volatile int fl_stop=0;
volatile int fl_msg=0;
volatile MQTTClient_deliveryToken deliveredtoken;
int cnt=0;
char url[200]="tcp://";
void exit_err(int err)
{
// printf("\nPress any key...");
// getchar();
exit(err);
}
void delivered(void *context, MQTTClient_deliveryToken dt)
{
// printf("Message with token value %d delivery confirmed\n", dt);
deliveredtoken = dt;
}
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
int i;
char* payloadptr;
char* mb_ptr;
// printf(" topic: %s\n", topicName);
while(fl_msg) Sleep(10);
// sprintf(out_msgbuf,"15,0,");
mb_ptr=out_msgbuf;
*mb_ptr++ ='1';
*mb_ptr++ ='5';
*mb_ptr++ =',';
*mb_ptr++ ='0';
*mb_ptr++ =',';
payloadptr = topicName;
while(1)
{
*mb_ptr++ =(*payloadptr++);
if (*payloadptr==0) break;
}
*mb_ptr++ =',';
payloadptr = message->payload;
for(i=0; i<message->payloadlen; i++)
{
*mb_ptr++ =(*payloadptr++);
}
*mb_ptr++ =0;
fl_msg=1;
// printf("%s\n", out_msgbuf);
MQTTClient_freeMessage(&message);
MQTTClient_free(topicName);
return 1;
}
void connlost(void *context, char *cause)
{
fl_connect=0;
}
#pragma warn (disable : 2134)
DWORD WINAPI ThreadFunc(LPVOID p)
{
int addrlen = sizeof(addr_in);
int nbytes;
char* ptr;
char topic[200];
while(1)
{
nbytes = recvfrom(fd_in, in_msgbuf, MSGBUFSIZE, 0, (struct sockaddr *) &addr_in, &addrlen);
if (nbytes > 0)
{
in_msgbuf[nbytes]=0; // добавим конец строки
if((in_msgbuf[0]=='m')&&(in_msgbuf[1]=='b')&&(in_msgbuf[2]=='g')&&(in_msgbuf[3]=='I')&&(in_msgbuf[4]=='G')&&(in_msgbuf[5]=='a')) // проверим сигнатуру "mbgIGa"
{
if(strstr(in_msgbuf,"mbgIGaSTOP") !=NULL) // проверим команду остановки сервиса "stop"
{
fl_stop=1;
sprintf(in_msgbuf, "mbgIGa0");
}
else
{
memmove(in_msgbuf, &in_msgbuf[6], nbytes-6+1); // Удалим сигнатуру "mbgIGa"
ptr = strstr(in_msgbuf,",");
if(ptr !=NULL)
{
*ptr=0; // заменим символ "," на 0
memcpy(topic, in_msgbuf, ptr-in_msgbuf+1);
//printf("%s\n", topic);
memmove(in_msgbuf, &in_msgbuf[ptr-in_msgbuf+1], nbytes-4+1-(ptr-in_msgbuf+1)); //
//printf("%s\n", in_msgbuf);
if (fl_connect!=0)
{
MQTTClient_publish(client, topic, strlen(in_msgbuf),in_msgbuf, 0, 0, NULL);
sprintf(in_msgbuf, "mbgIGa0");
}
else sprintf(in_msgbuf, "mbgIGa1"); // ошибка "нет связи"
}
else sprintf(in_msgbuf, "mbgIGa2"); // ошибка "некорректный топик"(нет разделителя ",")
}
nbytes = sendto( fd_in, in_msgbuf, strlen(in_msgbuf), 0, (struct sockaddr*) &addr_in, addrlen);
}
}
}
}
int main(int argc, char* argv[])
{
int i;
FreeConsole();
// printf("%s\n",argv[0]); // exe
// printf("%s\n",argv[1]); // сервер:порт
// printf("%s\n",argv[2]); // топик
// printf("%s\n",argv[3]); // ID
// printf("%s\n",argv[4]); // login
// printf("%s\n",argv[5]); // пароль
// printf("\nPress any key...");
// getchar();
/////////////////////////// UDP init /////////////////////////////////////////////////
WSADATA wsaData;
if (WSAStartup(0x0101, &wsaData)) {
perror("Error WSAStartup");
exit_err(1);
}
fd_in = socket(AF_INET, SOCK_DGRAM, 0); // in socket
if (fd_in < 0) {
perror("Error IN socket");
exit_err(2);
}
int fd_out = socket(AF_INET, SOCK_DGRAM, 0); // out socket
if (fd_out < 0) {
perror("Error OUT socket");
exit_err(3);
}
u_int yes = 1;
if (setsockopt(fd_in, SOL_SOCKET, SO_REUSEADDR, (char*) &yes, sizeof(yes)) < 0)
{
perror("Reusing ADDR failed");
exit_err(4);
}
memset(&addr_in, 0, sizeof(addr_in));
addr_in.sin_family = AF_INET;
addr_in.sin_addr.s_addr = htonl(INADDR_ANY); // differs from sender
addr_in.sin_port = htons(IN_PORT);
// bind to receive address
if (bind(fd_in, (struct sockaddr*) &addr_in, sizeof(addr_in)) < 0) {
perror("bind");
exit_err(5);
}
memset(&addr_out, 0, sizeof(addr_out));
addr_out.sin_family = AF_INET;
addr_out.sin_addr.s_addr = inet_addr("127.0.0.1");
addr_out.sin_port = htons(PORT_1C);
HANDLE hThread;
DWORD threadld;
hThread = CreateThread(NULL, 0, ThreadFunc, 0, 0, &threadld );
if (hThread==NULL) {
perror("Create thread error");
exit_err(6);
}
///////////////////////////////////////////////////////////////////////////////
HANDLE Test_Present = CreateMutex(NULL,TRUE,"mbsvs_Already_Present"); // защита от повторного запуска
if(GetLastError() == ERROR_ALREADY_EXISTS)
{
//printf("MQService already running!\n");
exit_err(7);
}
if(argc<6) // если в командной строке меньше 6 аргументов
{
//printf("Bad arguments!\n");
exit_err(8);
}
strcat(url,argv[1]); // full URL
// printf("client_id: %s\n ", url);
MQTTClient_create(&client, url, argv[3], MQTTCLIENT_PERSISTENCE_NONE, NULL);
conn_opts.keepAliveInterval = 60;
conn_opts.cleansession = 1;
conn_opts.username = argv[4];
conn_opts.password = argv[5];
conn_opts.connectTimeout = 5;
MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
fl_connect=0;
sprintf(out_msgbuf,"15,4,service started");
sendto(fd_out, out_msgbuf , strlen(out_msgbuf ),0,(struct sockaddr*) &addr_out, sizeof(addr_out));
while(1)
{
if(fl_connect==0)
{
while(1)
{
//printf("Connect to server...\n");
if (MQTTClient_connect(client, &conn_opts) == MQTTCLIENT_SUCCESS) break;
//printf("Connection error!\n");
sprintf(out_msgbuf,"15,1,server not connected");
sendto(fd_out, out_msgbuf , strlen(out_msgbuf ),0,(struct sockaddr*) &addr_out, sizeof(addr_out));
for (i=0;i<100;i++)
{
if(fl_stop) goto exit;
Sleep(50);
}
}
MQTTClient_subscribe(client, argv[2], QOS);
//printf("Subscribed topic: %s\n", argv[2]);
fl_connect=1;
sprintf(out_msgbuf,"15,2,server connected");
sendto(fd_out, out_msgbuf , strlen(out_msgbuf ),0,(struct sockaddr*) &addr_out, sizeof(addr_out));
}
else
{
if (fl_msg)
{
sendto(fd_out, out_msgbuf , strlen(out_msgbuf),0,(struct sockaddr*) &addr_out, sizeof(addr_out));
fl_msg=0;
}
else Sleep(10);
}
if(fl_stop) break;
}
exit:
sprintf(out_msgbuf,"15,3,service stopped");
sendto(fd_out, out_msgbuf , strlen(out_msgbuf ),0,(struct sockaddr*) &addr_out, sizeof(addr_out));
MQTTClient_disconnect(client, 10000);
MQTTClient_destroy(&client);
ReleaseMutex(Test_Present);
CloseHandle(Test_Present);
WSACleanup();
return (0);
}