GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-17 04:40:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2012 from fengzeroz/v2.4

Browse Source 
test
This commit is contained in:
fengzeroz
2024-05-21 10:16:39 +08:00
committed by GitHub
parent 29b10bdd31 06e50c174b
commit 3ff8ad64da
10 changed files with 637 additions and 255 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/connection/mqtt_client.h
+2 -2
View File
@@ -78,8 +78,8 @@ int neu_mqtt_client_set_connect_cb(neu_mqtt_client_t * client,
int neu_mqtt_client_set_disconnect_cb(neu_mqtt_client_t * client,
neu_mqtt_client_connection_cb_t cb,
void * data);
int neu_mqtt_client_set_tls(neu_mqtt_client_t *client, const char *ca,
const char *cert, const char *key,
int neu_mqtt_client_set_tls(neu_mqtt_client_t *client, bool enabled,
const char *ca, const char *cert, const char *key,
const char *keypass);
int neu_mqtt_client_set_cache_size(neu_mqtt_client_t *client,
size_t mem_size_bytes, size_t db_size_bytes);
include/neuron/connection/neu_connection.h
+11
View File
@@ -30,6 +30,7 @@ typedef enum neu_conn_type {
NEU_CONN_TCP_CLIENT,
NEU_CONN_UDP,
NEU_CONN_TTY_CLIENT,
NEU_CONN_UDP_TO,
} neu_conn_type_e;
typedef enum neu_conn_tty_baud {
@@ -103,6 +104,12 @@ typedef struct neu_conn_param {
uint16_t timeout; // millisecond
} udp;
struct {
char * src_ip;
uint16_t src_port;
uint16_t timeout; // millisecond
} udpto;
struct {
char * device;
neu_conn_tty_data_e data;
@@ -231,6 +238,8 @@ int neu_conn_tcp_server_close_client(neu_conn_t *conn, int fd);
* less than or equal to 0 fails.
*/
ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len);
ssize_t neu_conn_udp_sendto(neu_conn_t *conn, uint8_t *buf, ssize_t len,
void *dst);
/**
* @brief Receive data via connection read.
@@ -242,6 +251,8 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len);
* received, less than or equal to 0 fails.
*/
ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len);
ssize_t neu_conn_udp_recvfrom(neu_conn_t *conn, uint8_t *buf, ssize_t len,
void *src);
/**
* @brief Specify the client to send data.
include/neuron/connection/neu_connection_eth.h
+4 -8
View File
@@ -32,23 +32,19 @@ neu_conn_eth_t *neu_conn_eth_init(const char *interface, void *ctx);
int neu_conn_eth_uninit(neu_conn_eth_t *conn);
int neu_conn_eth_check_interface(const char *interface);
void neu_conn_eth_get_mac(neu_conn_eth_t *conn, uint8_t *mac);
int neu_conn_eth_size();
typedef void (*neu_conn_eth_msg_callback)(neu_conn_eth_t *conn, void *ctx,
uint16_t protocol, uint16_t n_byte,
uint8_t *bytes, uint8_t src_mac[6]);
int neu_conn_eth_register_lldp(neu_conn_eth_t *conn, const char *device,
neu_conn_eth_msg_callback callback);
int neu_conn_eth_unregister_lldp(neu_conn_eth_t *conn, const char *device);
typedef struct neu_conn_eth_sub neu_conn_eth_sub_t;
neu_conn_eth_sub_t *neu_conn_eth_register(neu_conn_eth_t *conn,
uint16_t protocol, uint8_t mac[6],
neu_conn_eth_sub_t *neu_conn_eth_register(neu_conn_eth_t *conn, uint8_t mac[6],
neu_conn_eth_msg_callback callback);
int neu_conn_eth_unregister(neu_conn_eth_t *conn, neu_conn_eth_sub_t *sub);
int neu_conn_eth_send(neu_conn_eth_t *conn, uint8_t dst_mac[6],
uint16_t protocol, uint16_t n_byte, uint8_t *bytes);
int neu_conn_eth_send(neu_conn_eth_t *conn, uint16_t protocol, uint16_t n_byte,
uint8_t *bytes);
#endif
include/neuron/connection/neu_smart_link.h
+25
View File
@@ -0,0 +1,25 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2023 EMQ Technologies Co., Ltd All rights reserved.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 3 of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this program; if not, write to the Free Software Foundation,
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#ifndef NEURON_CONNECTION_SMART_LINK_H
#define NEURON_CONNECTION_SMART_LINK_H
int neu_conn_smart_link_auto_set(const char *dev_path);
#endif
plugins/monitor/monitor.c
+2 -2
View File
@@ -151,8 +151,8 @@ static int config_mqtt_client(neu_plugin_t *plugin, neu_mqtt_client_t *client,
}
// when config->ca is NULL tls will be disabled
rv = neu_mqtt_client_set_tls(client, config->ca, config->cert, config->key,
config->keypass);
rv = neu_mqtt_client_set_tls(client, true, config->ca, config->cert,
config->key, config->keypass);
if (0 != rv) {
plog_error(plugin, "neu_mqtt_client_set_tsl fail");
return -1;
plugins/mqtt/mqtt_plugin.c
+2 -2
View File
@@ -146,8 +146,8 @@ static int config_mqtt_client(neu_plugin_t *plugin, neu_mqtt_client_t *client,
}
// when config->ca is NULL tls will be disabled
rv = neu_mqtt_client_set_tls(client, config->ca, config->cert, config->key,
config->keypass);
rv = neu_mqtt_client_set_tls(client, true, config->ca, config->cert,
config->key, config->keypass);
if (0 != rv) {
plog_error(plugin, "neu_mqtt_client_set_tsl fail");
return -1;
src/connection/connection.c
+175 -30
View File
@@ -319,6 +319,7 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len)
if (conn->is_connected) {
switch (conn->param.type) {
case NEU_CONN_UDP_TO:
case NEU_CONN_TCP_SERVER:
assert(false);
break;
@@ -378,6 +379,7 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
}
switch (conn->param.type) {
case NEU_CONN_UDP_TO:
case NEU_CONN_TCP_SERVER:
zlog_fatal(conn->param.log, "neu_conn_recv cann't recv tcp server msg");
assert(1 == 0);
@@ -432,6 +434,109 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
return ret;
}
ssize_t neu_conn_udp_sendto(neu_conn_t *conn, uint8_t *buf, ssize_t len,
void *dst)
{
ssize_t ret = 0;
pthread_mutex_lock(&conn->mtx);
if (conn->stop) {
pthread_mutex_unlock(&conn->mtx);
return ret;
}
if (!conn->is_connected) {
conn_connect(conn);
}
if (conn->is_connected) {
switch (conn->param.type) {
case NEU_CONN_TCP_CLIENT:
case NEU_CONN_TCP_SERVER:
case NEU_CONN_TTY_CLIENT:
case NEU_CONN_UDP:
assert(false);
break;
case NEU_CONN_UDP_TO:
if (conn->block) {
ret = sendto(conn->fd, buf, len, MSG_NOSIGNAL,
(struct sockaddr *) dst, sizeof(struct sockaddr));
} else {
ret = sendto(conn->fd, buf, len, MSG_NOSIGNAL | MSG_DONTWAIT,
(struct sockaddr *) dst, sizeof(struct sockaddr));
}
break;
}
if (ret != len) {
zlog_error(
conn->param.log,
"conn udp fd: %d, sendto (%s:%d) buf len: %zd, ret: %zd, "
"errno: %s(%d)",
conn->fd, inet_ntoa(((struct sockaddr_in *) dst)->sin_addr),
htons(((struct sockaddr_in *) dst)->sin_port), len, ret,
strerror(errno), errno);
}
if (ret == -1 && errno != EAGAIN) {
conn_disconnect(conn);
}
if (ret > 0 && conn->callback_trigger == false) {
conn->connected(conn->data, conn->fd);
conn->callback_trigger = true;
}
}
if (ret > 0) {
conn->state.send_bytes += ret;
}
pthread_mutex_unlock(&conn->mtx);
return ret;
}
ssize_t neu_conn_udp_recvfrom(neu_conn_t *conn, uint8_t *buf, ssize_t len,
void *src)
{
ssize_t ret = 0;
socklen_t addr_len = sizeof(struct sockaddr_in);
pthread_mutex_lock(&conn->mtx);
if (conn->stop) {
pthread_mutex_unlock(&conn->mtx);
return ret;
}
switch (conn->param.type) {
case NEU_CONN_TCP_SERVER:
case NEU_CONN_TCP_CLIENT:
case NEU_CONN_TTY_CLIENT:
case NEU_CONN_UDP:
assert(1 == 0);
break;
case NEU_CONN_UDP_TO:
ret =
recvfrom(conn->fd, buf, len, 0, (struct sockaddr *) src, &addr_len);
break;
}
if (ret <= 0) {
zlog_error(conn->param.log,
"conn udp fd: %d, recv buf len %zd, ret: %zd, errno: %s(%d)",
conn->fd, len, ret, strerror(errno), errno);
if (ret == 0 || (ret == -1 && errno != EAGAIN)) {
conn_disconnect(conn);
}
}
pthread_mutex_unlock(&conn->mtx);
if (ret > 0) {
conn->state.recv_bytes += ret;
}
return ret;
}
void neu_conn_connect(neu_conn_t *conn)
{
@@ -471,6 +576,9 @@ static void conn_free_param(neu_conn_t *conn)
free(conn->param.params.udp.src_ip);
free(conn->param.params.udp.dst_ip);
break;
case NEU_CONN_UDP_TO:
free(conn->param.params.udpto.src_ip);
break;
case NEU_CONN_TTY_CLIENT:
free(conn->param.params.tty_client.device);
break;
@@ -501,10 +609,6 @@ static void conn_init_param(neu_conn_t *conn, neu_conn_param_t *param)
} else {
conn->block = false;
}
zlog_notice(conn->param.log, "tcp server %s:%d, timeout: %d, block: %d",
conn->param.params.tcp_server.ip,
conn->param.params.tcp_server.port,
conn->param.params.tcp_server.timeout, conn->block);
break;
case NEU_CONN_TCP_CLIENT:
conn->param.params.tcp_client.ip = strdup(param->params.tcp_client.ip);
@@ -512,10 +616,6 @@ static void conn_init_param(neu_conn_t *conn, neu_conn_param_t *param)
conn->param.params.tcp_client.timeout =
param->params.tcp_client.timeout;
conn->block = conn->param.params.tcp_client.timeout > 0;
zlog_notice(conn->param.log, "tcp client %s:%d, timeout: %d, block: %d",
conn->param.params.tcp_client.ip,
conn->param.params.tcp_client.port,
conn->param.params.tcp_client.timeout, conn->block);
break;
case NEU_CONN_UDP:
conn->param.params.udp.src_ip = strdup(param->params.udp.src_ip);
@@ -524,11 +624,12 @@ static void conn_init_param(neu_conn_t *conn, neu_conn_param_t *param)
conn->param.params.udp.dst_port = param->params.udp.dst_port;
conn->param.params.udp.timeout = param->params.udp.timeout;
conn->block = conn->param.params.udp.timeout > 0;
zlog_notice(
conn->param.log, "udp %s:%d -> %s:%d, timeout: %d, block: %d",
conn->param.params.udp.src_ip, conn->param.params.udp.src_port,
conn->param.params.udp.dst_ip, conn->param.params.udp.dst_port,
conn->param.params.udp.timeout, conn->block);
break;
case NEU_CONN_UDP_TO:
conn->param.params.udpto.src_ip = strdup(param->params.udpto.src_ip);
conn->param.params.udpto.src_port = param->params.udpto.src_port;
conn->param.params.udpto.timeout = param->params.udpto.timeout;
conn->block = conn->param.params.udpto.timeout > 0;
break;
case NEU_CONN_TTY_CLIENT:
conn->param.params.tty_client.device =
@@ -651,9 +752,9 @@ static void conn_connect(neu_conn_t *conn)
conn->is_connected = false;
return;
} else {
zlog_notice(conn->param.log, "connect %s:%d, block: %d success",
zlog_notice(conn->param.log, "connect %s:%d success",
conn->param.params.tcp_client.ip,
conn->param.params.tcp_client.port, conn->block);
conn->param.params.tcp_client.port);
conn->is_connected = true;
conn->fd = fd;
}
@@ -714,10 +815,57 @@ static void conn_connect(neu_conn_t *conn)
}
break;
}
case NEU_CONN_UDP_TO: {
if (conn->block) {
struct timeval tv = {
.tv_sec = conn->param.params.udpto.timeout / 1000,
.tv_usec = (conn->param.params.udpto.timeout % 1000) * 1000,
};
fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
} else {
fd = socket(AF_INET, SOCK_DGRAM | SOCK_NONBLOCK, IPPROTO_UDP);
}
int so_broadcast = 1;
setsockopt(fd, SOL_SOCKET, SO_BROADCAST, &so_broadcast,
sizeof(so_broadcast));
struct sockaddr_in local = {
.sin_family = AF_INET,
.sin_port = htons(conn->param.params.udpto.src_port),
.sin_addr.s_addr = inet_addr(conn->param.params.udpto.src_ip),
};
ret = bind(fd, (struct sockaddr *) &local, sizeof(struct sockaddr_in));
if (ret != 0) {
close(fd);
zlog_error(conn->param.log, "bind %s:%d error: %s(%d)",
conn->param.params.udpto.src_ip,
conn->param.params.udpto.src_port, strerror(errno),
errno);
conn->is_connected = false;
return;
}
conn->is_connected = true;
conn->fd = fd;
break;
}
case NEU_CONN_TTY_CLIENT: {
struct termios tty_opt = { 0 };
#ifdef NEU_SMART_LINK
#include "connection/neu_smart_link.h"
ret =
neu_conn_smart_link_auto_set(conn->param.params.tty_client.device);
zlog_notice(conn->param.log, "smart link ret: %d", ret);
if (ret > 0) {
fd = ret;
}
#else
fd = open(conn->param.params.tty_client.device, O_RDWR | O_NOCTTY, 0);
#endif
if (fd <= 0) {
zlog_error(conn->param.log, "open %s error: %s(%d)",
conn->param.params.tty_client.device, strerror(errno),
@@ -894,6 +1042,7 @@ static void conn_disconnect(neu_conn_t *conn)
break;
case NEU_CONN_TCP_CLIENT:
case NEU_CONN_UDP:
case NEU_CONN_UDP_TO:
case NEU_CONN_TTY_CLIENT:
if (conn->fd > 0) {
close(conn->fd);
@@ -970,15 +1119,13 @@ int neu_conn_stream_consume(neu_conn_t *conn, void *context,
} else if (used == -1) {
neu_conn_disconnect(conn);
break;
} else {
conn->offset -= used;
memmove(conn->buf, conn->buf + used, conn->offset);
neu_protocol_unpack_buf_init(&protocol_buf, conn->buf,
conn->offset);
}
}
if (conn->offset != 0) {
conn->offset -= neu_protocol_unpack_buf_used_size(&protocol_buf);
memmove(conn->buf,
conn->buf +
neu_protocol_unpack_buf_used_size(&protocol_buf),
conn->offset);
}
}
return ret;
@@ -1001,15 +1148,13 @@ int neu_conn_stream_tcp_server_consume(neu_conn_t *conn, int fd, void *context,
} else if (used == -1) {
neu_conn_tcp_server_close_client(conn, fd);
break;
} else {
conn->offset -= used;
memmove(conn->buf, conn->buf + used, conn->offset);
neu_protocol_unpack_buf_init(&protocol_buf, conn->buf,
conn->offset);
}
}
if (conn->offset != 0) {
conn->offset -= neu_protocol_unpack_buf_used_size(&protocol_buf);
memmove(conn->buf,
conn->buf +
neu_protocol_unpack_buf_used_size(&protocol_buf),
conn->offset);
}
}
return ret;
}
src/connection/connection_eth.c
+114 -207
View File
@@ -17,6 +17,7 @@
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#include <assert.h>
#include <errno.h>
#include <pthread.h>
#include <stdbool.h>
#include <stdio.h>
@@ -32,6 +33,7 @@
#include "connection/neu_connection_eth.h"
#include "event/event.h"
#include "type.h"
#include "utils/log.h"
#include "utils/utextend.h"
@@ -39,13 +41,12 @@
#define ETH_P_LLDP 0x88CC
#endif
typedef struct neu_conn_eth_sub {
uint16_t protocol;
char mac[ETH_ALEN];
} callback_elem_key_t;
struct neu_conn_eth_sub {
uint8_t mac[6];
};
typedef struct {
callback_elem_key_t key;
char mac[20];
neu_conn_eth_msg_callback callback;
@@ -53,8 +54,6 @@ typedef struct {
} callback_elem_t;
typedef struct {
char device[NEU_NODE_NAME_LEN];
neu_conn_eth_msg_callback callback;
UT_hash_handle hh;
@@ -64,20 +63,16 @@ typedef struct interface_conn {
char * interface;
neu_events_t *events;
uint8_t count;
uint8_t buf[1500];
int lldp_fd;
int profinet_fd;
int vlan_fd;
uint8_t mac[ETH_ALEN];
neu_event_io_t *profinet_event;
neu_event_io_t *lldp_event;
neu_event_io_t *vlan_event;
lldp_callback_elem_t *lldp_callbacks;
callback_elem_t * callbacks;
pthread_mutex_t mtx;
callback_elem_t *callbacks;
pthread_mutex_t mtx;
} interface_conn_t;
static pthread_mutex_t mtx = { 0 };
@@ -91,13 +86,25 @@ struct neu_conn_eth {
};
static int get_mac(neu_conn_eth_t *conn, const char *interface);
static int init_socket(const char *interface, uint16_t protocol);
static int init_socket(const char *interface, uint16_t protocol,
uint8_t mac[6]);
static int uninit_socket(int fd);
static int lldp_msg_cb(enum neu_event_io_type type, int fd, void *usr_data);
static int eth_msg_cb(enum neu_event_io_type type, int fd, void *usr_data);
static int eth_msg_cb(enum neu_event_io_type type, int fd, void *usr_data);
static uint8_t pf_dcp_broadcast[ETH_ALEN] = {
0x01, 0x0e, 0xcf, 0x00, 0x00, 0x01
};
static char *lldp_parse_port_id(uint16_t n_byte, uint8_t *bytes);
int neu_conn_eth_size()
{
int ret = 0;
for (int i = 0; i < interface_max; i++) {
ret += in_conns[i].count;
}
return ret;
}
neu_conn_eth_t *neu_conn_eth_init(const char *interface, void *ctx)
{
@@ -126,17 +133,19 @@ neu_conn_eth_t *neu_conn_eth_init(const char *interface, void *ctx)
if (in_conns[i].interface == NULL) {
neu_event_io_param_t param = { 0 };
pthread_mutex_init(&in_conns[i].mtx, NULL);
conn_eth->ic = &in_conns[i];
in_conns[i].interface = strdup(interface);
in_conns[i].events = neu_event_new();
in_conns[i].count += 1;
in_conns[i].profinet_fd = init_socket(interface, 0x8892);
in_conns[i].lldp_fd = init_socket(interface, ETH_P_LLDP);
in_conns[i].vlan_fd = init_socket(interface, 0x8100);
get_mac(conn_eth, interface);
pthread_mutex_init(&in_conns[i].mtx, NULL);
conn_eth->ic = &in_conns[i];
in_conns[i].profinet_fd =
init_socket(interface, 0x8892, in_conns[i].mac);
// in_conns[i].vlan_fd =
// init_socket(interface, 0x8100, in_conns[i].mac);
param.fd = in_conns[i].profinet_fd;
param.usr_data = (void *) conn_eth;
@@ -145,17 +154,11 @@ neu_conn_eth_t *neu_conn_eth_init(const char *interface, void *ctx)
in_conns[i].profinet_event =
neu_event_add_io(in_conns[i].events, param);
param.fd = in_conns[i].lldp_fd;
param.cb = lldp_msg_cb;
// param.fd = in_conns[i].vlan_fd;
// param.cb = eth_msg_cb;
in_conns[i].lldp_event =
neu_event_add_io(in_conns[i].events, param);
param.fd = in_conns[i].vlan_fd;
param.cb = eth_msg_cb;
in_conns[i].vlan_event =
neu_event_add_io(in_conns[i].events, param);
// in_conns[i].vlan_event =
// neu_event_add_io(in_conns[i].events, param);
break;
}
@@ -174,16 +177,15 @@ int neu_conn_eth_uninit(neu_conn_eth_t *conn)
conn->ic->count -= 1;
if (conn->ic->count == 0) {
neu_event_del_io(conn->ic->events, conn->ic->profinet_event);
neu_event_del_io(conn->ic->events, conn->ic->lldp_event);
neu_event_del_io(conn->ic->events, conn->ic->vlan_event);
// neu_event_del_io(conn->ic->events, conn->ic->vlan_event);
neu_event_close(conn->ic->events);
uninit_socket(conn->ic->profinet_fd);
uninit_socket(conn->ic->lldp_fd);
uninit_socket(conn->ic->vlan_fd);
// uninit_socket(conn->ic->vlan_fd);
free(conn->ic->interface);
conn->ic->interface = NULL;
pthread_mutex_destroy(&conn->ic->mtx);
}
pthread_mutex_unlock(&mtx);
@@ -201,43 +203,46 @@ void neu_conn_eth_get_mac(neu_conn_eth_t *conn, uint8_t *mac)
int neu_conn_eth_check_interface(const char *interface)
{
struct ifreq ifr = { 0 };
int fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_IP);
int fd = socket(PF_PACKET, SOCK_RAW, ETH_P_IPV6);
int ret = -1;
if (fd <= 0) {
return -1;
}
strcpy(ifr.ifr_name, interface);
ret = ioctl(fd, SIOCGIFHWADDR, &ifr);
if (ret != 0) {
return -1;
}
close(fd);
return ret;
}
neu_conn_eth_sub_t *neu_conn_eth_register(neu_conn_eth_t *conn,
uint16_t protocol, uint8_t mac[6],
neu_conn_eth_sub_t *neu_conn_eth_register(neu_conn_eth_t *conn, uint8_t xmac[6],
neu_conn_eth_msg_callback callback)
{
callback_elem_key_t key = { 0 };
neu_conn_eth_sub_t *sub = NULL;
callback_elem_t * elem = NULL;
char tmp_mac[20] = { 0 };
neu_conn_eth_sub_t *sub = NULL;
callback_elem_t * elem = NULL;
key.protocol = protocol;
memcpy(key.mac, mac, ETH_ALEN);
snprintf(tmp_mac, sizeof(tmp_mac), "%X:%X:%X-%X:%X:%X", xmac[0], xmac[1],
xmac[2], xmac[3], xmac[4], xmac[5]);
pthread_mutex_lock(&conn->ic->mtx);
HASH_FIND(hh, conn->ic->callbacks, &key, sizeof(callback_elem_key_t), elem);
HASH_FIND_STR(conn->ic->callbacks, tmp_mac, elem);
if (elem == NULL) {
elem = calloc(1, sizeof(callback_elem_t));
sub = calloc(1, sizeof(neu_conn_eth_sub_t));
elem->key.protocol = protocol;
memcpy(key.mac, mac, ETH_ALEN);
elem->callback = callback;
memcpy(elem->mac, tmp_mac, sizeof(elem->mac));
memcpy(sub->mac, mac, ETH_ALEN);
sub->protocol = protocol;
memcpy(sub->mac, xmac, ETH_ALEN);
HASH_ADD(hh, conn->ic->callbacks, key, sizeof(callback_elem_key_t),
elem);
HASH_ADD_STR(conn->ic->callbacks, mac, elem);
}
pthread_mutex_unlock(&conn->ic->mtx);
@@ -247,15 +252,15 @@ neu_conn_eth_sub_t *neu_conn_eth_register(neu_conn_eth_t *conn,
int neu_conn_eth_unregister(neu_conn_eth_t *conn, neu_conn_eth_sub_t *sub)
{
callback_elem_key_t key = { 0 };
callback_elem_t * elem = NULL;
char tmp_mac[20] = { 0 };
callback_elem_t *elem = NULL;
key.protocol = sub->protocol;
memcpy(key.mac, sub->mac, ETH_ALEN);
snprintf(tmp_mac, sizeof(tmp_mac), "%X:%X:%X-%X:%X:%X", sub->mac[0],
sub->mac[1], sub->mac[2], sub->mac[3], sub->mac[4], sub->mac[5]);
pthread_mutex_lock(&conn->ic->mtx);
HASH_FIND(hh, conn->ic->callbacks, &key, sizeof(callback_elem_key_t), elem);
HASH_FIND_STR(conn->ic->callbacks, tmp_mac, elem);
if (elem != NULL) {
HASH_DEL(conn->ic->callbacks, elem);
free(elem);
@@ -268,61 +273,17 @@ int neu_conn_eth_unregister(neu_conn_eth_t *conn, neu_conn_eth_sub_t *sub)
return 0;
}
int neu_conn_eth_register_lldp(neu_conn_eth_t *conn, const char *device,
neu_conn_eth_msg_callback callback)
int neu_conn_eth_send(neu_conn_eth_t *conn, uint16_t protocol, uint16_t n_byte,
uint8_t *bytes)
{
lldp_callback_elem_t *elem = NULL;
int ret = 0;
pthread_mutex_lock(&conn->ic->mtx);
HASH_FIND_STR(conn->ic->lldp_callbacks, device, elem);
if (elem != NULL) {
ret = -1;
} else {
elem = calloc(1, sizeof(lldp_callback_elem_t));
elem->callback = callback;
strcpy(elem->device, device);
HASH_ADD_STR(conn->ic->lldp_callbacks, device, elem);
}
pthread_mutex_unlock(&conn->ic->mtx);
return ret;
}
int neu_conn_eth_unregister_lldp(neu_conn_eth_t *conn, const char *device)
{
lldp_callback_elem_t *elem = NULL;
pthread_mutex_lock(&conn->ic->mtx);
HASH_FIND_STR(conn->ic->lldp_callbacks, device, elem);
if (elem != NULL) {
HASH_DEL(conn->ic->lldp_callbacks, elem);
free(elem);
}
pthread_mutex_unlock(&conn->ic->mtx);
return 0;
}
int neu_conn_eth_send(neu_conn_eth_t *conn, uint8_t dst_mac[6],
uint16_t protocol, uint16_t n_byte, uint8_t *bytes)
{
int fd = 0;
int ret = 0;
struct ethhdr *ehdr = (struct ethhdr *) bytes;
(void) dst_mac;
int fd = 0;
int ret = 0;
switch (protocol) {
case ETH_P_LLDP:
case 0x8892:
fd = conn->ic->profinet_fd;
break;
case ETH_P_LLDP:
fd = conn->ic->lldp_fd;
break;
case 0x8100:
fd = conn->ic->vlan_fd;
break;
@@ -331,8 +292,6 @@ int neu_conn_eth_send(neu_conn_eth_t *conn, uint8_t dst_mac[6],
break;
}
memcpy(ehdr->h_source, conn->ic->mac, ETH_ALEN);
ret = send(fd, bytes, n_byte, 0);
return ret;
}
@@ -353,20 +312,22 @@ static int get_mac(neu_conn_eth_t *conn, const char *interface)
return ret;
}
static int init_socket(const char *interface, uint16_t protocol)
static int init_socket(const char *interface, uint16_t protocol, uint8_t mac[6])
{
struct ifreq ifr = { 0 };
struct sockaddr_ll sll = { 0 };
struct packet_mreq mreq = { 0 };
const uint8_t pn_mcast_addr[ETH_ALEN] = {
0x01, 0x0e, 0xcf, 0x00, 0x00, 0x00
};
int fd = socket(PF_PACKET, SOCK_RAW, htons(protocol));
int ret = 0;
int ret = 1;
assert(fd > 0);
setsockopt(fd, SOL_SOCKET, SO_DONTROUTE, &ret, sizeof(ret));
(void) mac;
(void) pn_mcast_addr;
strcpy(ifr.ifr_name, interface);
if (ioctl(fd, SIOCGIFINDEX, &ifr) != 0) {
close(fd);
@@ -376,25 +337,12 @@ static int init_socket(const char *interface, uint16_t protocol)
sll.sll_family = AF_PACKET;
sll.sll_ifindex = ifr.ifr_ifindex;
sll.sll_protocol = htons(protocol);
mreq.mr_ifindex = ifr.ifr_ifindex;
mreq.mr_type = PACKET_HOST | PACKET_MR_MULTICAST;
mreq.mr_alen = ETH_ALEN;
memcpy(mreq.mr_address, pn_mcast_addr, ETH_ALEN);
ifr.ifr_flags = 0;
ioctl(fd, SIOCGIFFLAGS, &ifr);
bind(fd, (struct sockaddr *) &sll, sizeof(sll));
if (protocol == 0x8892) {
ret = setsockopt(fd, SOL_PACKET, PACKET_ADD_MEMBERSHIP, &mreq,
sizeof(mreq));
assert(ret == 0);
}
ret = bind(fd, (struct sockaddr *) &sll, sizeof(sll));
assert(ret == 0);
return fd;
}
static int uninit_socket(int fd)
{
close(fd);
@@ -408,27 +356,53 @@ static int eth_msg_cb(enum neu_event_io_type type, int fd, void *usr_data)
switch (type) {
case NEU_EVENT_IO_READ: {
callback_elem_key_t key = { 0 };
callback_elem_t * elem = NULL;
memset(conn->ic->buf, 0, sizeof(conn->ic->buf));
int ret = recv(fd, conn->ic->buf, sizeof(conn->ic->buf), 0);
callback_elem_t *elem = NULL;
uint8_t buf[1500] = { 0 };
int ret = recv(fd, buf, 1500, 0);
if (ret < ETH_HLEN) {
nlog_warn("eth recv length < %d msg", ETH_HLEN);
return 0;
}
struct ethhdr *ehdr = (struct ethhdr *) conn->ic->buf;
struct ethhdr *ehdr = (struct ethhdr *) buf;
ehdr->h_proto = ntohs(ehdr->h_proto);
key.protocol = ntohs(ehdr->h_proto);
memcpy(key.mac, ehdr->h_source, ETH_ALEN);
if (ehdr->h_proto != 0x8100 && ehdr->h_proto != 0x8892) {
return 0;
}
HASH_FIND(hh, conn->ic->callbacks, &key, sizeof(key), elem);
if (memcmp(ehdr->h_dest, pf_dcp_broadcast, ETH_ALEN) == 0) {
callback_elem_t *tmp = NULL;
if (elem != NULL) {
elem->callback(conn, conn->ctx, key.protocol, ret - ETH_HLEN,
conn->ic->buf + ETH_HLEN, ehdr->h_source);
HASH_ITER(hh, conn->ic->callbacks, elem, tmp)
{
elem->callback(conn, conn->ctx, ehdr->h_proto, ret - ETH_HLEN,
buf + ETH_HLEN, ehdr->h_source);
}
} else {
if (memcmp(ehdr->h_dest, conn->ic->mac, ETH_ALEN) != 0) {
return 0;
}
char tmp_mac[20] = { 0 };
snprintf(tmp_mac, sizeof(tmp_mac), "%X:%X:%X-%X:%X:%X",
ehdr->h_source[0], ehdr->h_source[1], ehdr->h_source[2],
ehdr->h_source[3], ehdr->h_source[4], ehdr->h_source[5]);
HASH_FIND_STR(conn->ic->callbacks, tmp_mac, elem);
if (elem != NULL) {
elem->callback(conn, conn->ctx, ehdr->h_proto, ret - ETH_HLEN,
buf + ETH_HLEN, ehdr->h_source);
} else {
memset(tmp_mac, 0, sizeof(tmp_mac));
snprintf(tmp_mac, sizeof(tmp_mac), "0:0:0-0:0:0");
HASH_FIND_STR(conn->ic->callbacks, tmp_mac, elem);
if (elem != NULL) {
elem->callback(conn, conn->ctx, ehdr->h_proto,
ret - ETH_HLEN, buf + ETH_HLEN,
ehdr->h_source);
}
}
}
break;
@@ -442,70 +416,3 @@ static int eth_msg_cb(enum neu_event_io_type type, int fd, void *usr_data)
return 0;
}
static int lldp_msg_cb(enum neu_event_io_type type, int fd, void *usr_data)
{
neu_conn_eth_t *conn = (neu_conn_eth_t *) usr_data;
int ret = 0;
char * port_id = NULL;
switch (type) {
case NEU_EVENT_IO_READ:
memset(conn->ic->buf, 0, sizeof(conn->ic->buf));
ret = recv(fd, conn->ic->buf, sizeof(conn->ic->buf), 0);
port_id = lldp_parse_port_id(ret - sizeof(struct ethhdr),
conn->ic->buf + sizeof(struct ethhdr));
if (port_id != NULL) {
lldp_callback_elem_t *elem = NULL;
struct ethhdr * ehdr = (struct ethhdr *) conn->ic->buf;
HASH_FIND_STR(conn->ic->callbacks, port_id, elem);
if (elem != NULL) {
elem->callback(
conn, conn->ctx, ETH_P_LLDP, ret - sizeof(struct ethhdr),
conn->ic->buf + sizeof(struct ethhdr), ehdr->h_source);
}
}
break;
case NEU_EVENT_IO_CLOSED:
case NEU_EVENT_IO_HUP:
nlog_warn("eth conn lldp error type: %d, fd: %d(%s)", type, fd,
conn->ic->interface);
break;
}
return 0;
}
struct lldp_tvl {
uint16_t type : 7;
uint16_t length : 9;
} __attribute__((packed));
#define END_OF_LLDPDU_TLV_TYPE 0x00
#define CHASSIS_ID_TLV_TYPE 0x01
#define PORT_ID_TLV_TYPE 0x02
#define TIME_TO_LIVE_TLV_TYPE 0x03
static char *lldp_parse_port_id(uint16_t n_byte, uint8_t *bytes)
{
struct lldp_tvl *tvl = (struct lldp_tvl *) bytes;
if (n_byte < sizeof(struct lldp_tvl) + tvl->length) {
return NULL;
}
if (tvl->type == PORT_ID_TLV_TYPE) {
static __thread char port_id[NEU_NODE_NAME_LEN] = { 0 };
uint8_t * data = (uint8_t *) &tvl[1];
strncpy(port_id, (char *) data + 1, tvl->length - 1);
return port_id;
} else {
return lldp_parse_port_id(
n_byte - sizeof(struct lldp_tvl) - tvl->length,
bytes + sizeof(struct lldp_tvl) + tvl->length);
}
}
src/connection/mqtt_client.c
+16 -4
View File
@@ -1005,8 +1005,8 @@ int neu_mqtt_client_set_disconnect_cb(neu_mqtt_client_t * client,
return 0;
}
int neu_mqtt_client_set_tls(neu_mqtt_client_t *client, const char *ca,
const char *cert, const char *key,
int neu_mqtt_client_set_tls(neu_mqtt_client_t *client, bool enabled,
const char *ca, const char *cert, const char *key,
const char *keypass)
{
int rv = 0;
@@ -1015,7 +1015,7 @@ int neu_mqtt_client_set_tls(neu_mqtt_client_t *client, const char *ca,
nng_mtx_lock(client->mtx);
return_failure_if_open();
if (NULL == ca) {
if (!enabled) {
// disable tls
log(debug, "tls disabled");
if (client->tls_cfg) {
@@ -1025,6 +1025,12 @@ int neu_mqtt_client_set_tls(neu_mqtt_client_t *client, const char *ca,
goto end;
}
if (NULL == client->host) {
log(error, "no client host");
rv = -1;
goto end;
}
if (client->tls_cfg) {
cfg = client->tls_cfg;
} else {
@@ -1037,6 +1043,12 @@ int neu_mqtt_client_set_tls(neu_mqtt_client_t *client, const char *ca,
client->tls_cfg = cfg;
}
if (0 != (rv = nng_tls_config_server_name(client->tls_cfg, client->host))) {
log(error, "nng_tls_config_server_name fail: %s", nng_strerror(rv));
rv = -1;
goto end;
}
if (cert != NULL && key != NULL) {
if ((rv = nng_tls_config_auth_mode(cfg, NNG_TLS_AUTH_MODE_REQUIRED)) !=
0) {
@@ -1057,7 +1069,7 @@ int neu_mqtt_client_set_tls(neu_mqtt_client_t *client, const char *ca,
}
}
if ((rv = nng_tls_config_ca_chain(cfg, ca, NULL)) != 0) {
if (ca && (rv = nng_tls_config_ca_chain(cfg, ca, NULL)) != 0) {
log(error, "nng_tls_config_ca_chain fail: %s", nng_strerror(rv));
rv = -1;
}
src/connection/smart_link.c
+286
View File
@@ -0,0 +1,286 @@
#include <errno.h>
#include <fcntl.h>
#include <linux/serial.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/ioctl.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <termios.h>
#include <unistd.h>
#include <jansson.h>
#include "json/json.h"
#include "connection/neu_smart_link.h"
#define DEVICE_CONF_PATH "/etc/config/device.conf"
#define SERIAL_CONF_PATH "/etc/config/serial.conf"
#define SMART_LINK_PATH "/dev/smartlink"
#define TYPE 'S'
#define UART_RS232 _IOWR(TYPE, 1, int)
#define UART_RS485 _IOWR(TYPE, 0, int)
enum SERIAL_MODE { SM_RS232, SM_RS422, SM_RS485 };
static int get_port(const char *path);
static int get_port_property(int port, int *mode, int *baudrate);
static int read_file(const char *path, char **buf);
static int set_serial_mode(const char *dev_path, int baudrate, int mode);
int neu_conn_smart_link_auto_set(const char *dev_path)
{
int mode = 0;
int baudrate = 0;
int port = get_port(dev_path);
if (port <= 0) {
return -1;
}
printf("get port %d\n", port);
int ret = get_port_property(port, &mode, &baudrate);
if (ret != 0) {
return -1;
}
printf("get port property %d %d\n", mode, baudrate);
ret = set_serial_mode(dev_path, baudrate, mode);
return ret;
}
static int read_file(const char *path, char **buf)
{
FILE *fp = NULL;
fp = fopen(path, "r");
if (fp == NULL) {
return -1;
}
fseek(fp, 0, SEEK_END);
int size = ftell(fp);
rewind(fp);
*buf = calloc(1, size + 1);
if (*buf == NULL) {
fclose(fp);
return -1;
}
int ret = fread(*buf, 1, size, fp);
if (ret != size) {
fclose(fp);
free(*buf);
return -1;
}
fclose(fp);
return ret;
}
static int get_port(const char *dev_path)
{
json_t * root = NULL;
json_error_t error;
char * buf = NULL;
int ret = read_file(DEVICE_CONF_PATH, &buf);
if (ret <= 0) {
return -1;
}
root = json_loads(buf, JSON_DECODE_ANY, &error);
if (root == NULL) {
free(buf);
return -1;
}
neu_json_elem_t e_tty = {
.name = "tty_name",
.t = NEU_JSON_OBJECT,
};
ret = neu_json_decode_value(root, &e_tty);
if (ret != 0 || json_is_array((json_t *) e_tty.v.val_object) == false) {
free(buf);
json_decref(root);
return -1;
}
int index = 0;
json_t *e_dev = NULL;
json_array_foreach(e_tty.v.val_object, index, e_dev)
{
neu_json_elem_t e_dd = {
.name = "device",
.t = NEU_JSON_STR,
};
ret = neu_json_decode_value(e_dev, &e_dd);
if (ret != 0) {
free(buf);
json_decref(root);
return -1;
}
if (strcmp(e_dd.v.val_str, dev_path) == 0) {
neu_json_elem_t e_port = {
.name = "port",
.t = NEU_JSON_INT,
};
ret = neu_json_decode_value(e_dev, &e_port);
if (ret != 0) {
free(buf);
json_decref(root);
return -1;
}
free(buf);
json_decref(root);
free(e_dd.v.val_str);
return e_port.v.val_int;
}
free(e_dd.v.val_str);
}
free(buf);
json_decref(root);
return -1;
}
static int get_port_property(int port, int *mode, int *baudrate)
{
json_t * root = NULL;
json_error_t error;
char * buf = NULL;
int ret = read_file(SERIAL_CONF_PATH, &buf);
if (ret <= 0) {
return -1;
}
root = json_loads(buf, JSON_DECODE_ANY, &error);
if (root == NULL) {
free(buf);
return -1;
}
json_t *e_port = json_array_get(root, port - 1);
if (e_port == NULL) {
json_decref(root);
free(buf);
return -1;
}
neu_json_elem_t elems[] = {
{
.name = "mode",
.t = NEU_JSON_INT,
},
{
.name = "baudrate",
.t = NEU_JSON_INT,
},
};
ret = neu_json_decode_by_json(e_port, NEU_JSON_ELEM_SIZE(elems), elems);
if (ret != 0) {
free(buf);
json_decref(root);
return -1;
}
*mode = elems[0].v.val_int;
*baudrate = elems[1].v.val_int;
json_decref(root);
free(buf);
return 0;
}
static int set_serial_mode(const char *dev_path, int baudrate, int mode)
{
static pthread_mutex_t mode_switch_mtx = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_lock(&mode_switch_mtx);
int serial_fd, switch_fd;
int tty_num, para;
int ret = -1;
struct serial_rs485 conf;
conf.flags &= !SER_RS485_ENABLED;
switch (baudrate) {
case 38400:
conf.delay_rts_after_send = 1;
break;
case 57600:
conf.delay_rts_after_send = 4;
break;
case 115200:
conf.delay_rts_after_send = 6;
break;
default:
conf.delay_rts_after_send = 0;
break;
}
if ((serial_fd = open(dev_path, O_RDWR | O_NOCTTY)) < 0) {
fprintf(stderr, "open %s failed, fd: %d, strerr: %s\n", dev_path,
serial_fd, strerror(errno));
goto out;
}
if (mode != SM_RS232) {
if (ioctl(serial_fd, TIOCMGET, &para) != 0) {
fprintf(stderr, "ioctl failed, strerr: %s\n", strerror(errno));
goto out2;
}
if (mode == SM_RS422) {
para |= TIOCM_DTR;
conf.flags |= SER_RS485_ENABLED | SER_RS485_RTS_ON_SEND;
} else if (mode == SM_RS485) {
para &= ~TIOCM_DTR;
conf.flags |= SER_RS485_ENABLED | SER_RS485_RTS_ON_SEND;
}
if (ioctl(serial_fd, TIOCMSET, &para) != 0) {
fprintf(stderr, "ioctl failed, strerr: %s\n", strerror(errno));
goto out2;
}
}
ioctl(serial_fd, TIOCSRS485, &conf);
if ((switch_fd = open(SMART_LINK_PATH, O_RDWR)) < 0) {
fprintf(stderr, "open %s failed, ret: %d, strerr: %s\n",
SMART_LINK_PATH, switch_fd, strerror(errno));
goto out2;
}
sscanf(dev_path, "%*[a-zA-Z/]%d", &tty_num);
printf("tty_num: %d\n", tty_num);
if (mode == SM_RS232) {
ret = ioctl(switch_fd, UART_RS232, tty_num);
} else {
ret = ioctl(switch_fd, UART_RS485, tty_num);
}
if (ret < 0) {
fprintf(stderr, "ioctl failed, ret: %d, strerr: %s\n", ret,
strerror(errno));
}
close(switch_fd);
ret = serial_fd;
pthread_mutex_unlock(&mode_switch_mtx);
return ret;
out2:
close(serial_fd);
serial_fd = -1;
out:
pthread_mutex_unlock(&mode_switch_mtx);
return ret;
}