This commit is contained in:
Wolfgang Hottgenroth 2021-02-10 12:04:21 +01:00
parent 6b36d97c05
commit 869caa8466
2 changed files with 83 additions and 29 deletions

View File

@ -15,42 +15,61 @@
config_t cfg; config_t cfg;
config_setting_t *devicesConfig;
int receiveSockFd; typedef struct {
config_setting_t *devicesConfig;
int receiveSockFd;
} t_receiverHandle;
t_receiverHandle receiveHandle;
typedef struct {
char *influxUser;
char *influxPass;
char *influxServer;
uint16_t influxPort;
char *influxDatabase;
char *influxMeasurement;
} t_forwarderHandle;
t_forwarderHandle forwarderHandle = {
.influxUser = NULL, .influxPass = NULL, .influxServer = NULL,
.influxPort = 8086, .influxDatabase = NULL, .influxMeasurement = NULL
};
void readConfig() { int readConfig(config_t *cfg) {
config_init(&cfg); config_init(cfg);
if (! config_read_file(&cfg, "./sink20169.cfg")) { if (! config_read_file(cfg, "./sink20169.cfg")) {
logmsg(LOG_ERR, "failed to read config file: %s:%d - %s\n", logmsg(LOG_ERR, "failed to read config file: %s:%d - %s\n",
config_error_file(&cfg), config_error_line(&cfg), config_error_file(cfg), config_error_line(cfg),
config_error_text(&cfg)); config_error_text(cfg));
config_destroy(&cfg); config_destroy(cfg);
exit(-1); return -1;
} }
return 0;
}
int initReceiver(config_t *cfg, t_receiverHandle *handle) {
devicesConfig = config_lookup(&cfg, "devices"); devicesConfig = config_lookup(&cfg, "devices");
if (devicesConfig == NULL) { if (devicesConfig == NULL) {
logmsg(LOG_ERR, "receiver: no devices configuration found"); logmsg(LOG_ERR, "receiver: no devices configuration found");
exit(-2); exit(-2);
} }
}
void initReceiver() {
struct sockaddr_in servaddr; struct sockaddr_in servaddr;
receiveSockFd = socket(AF_INET, SOCK_DGRAM, 0); handle->receiveSockFd = socket(AF_INET, SOCK_DGRAM, 0);
if (receiveSockFd == -1) { if (handle->receiveSockFd == -1) {
logmsg(LOG_ERR, "failed to create receive socket: %d", errno); logmsg(LOG_ERR, "failed to create receive socket: %d", errno);
exit(-3); return -1;
} }
int receivePort = 20169; int receivePort = 20169;
config_lookup_int(&cfg, "receivePort", &receivePort); config_lookup_int(cfg, "receivePort", &receivePort);
if (receivePort < 1 || receivePort > 65535) { if (receivePort < 1 || receivePort > 65535) {
logmsg(LOG_ERR, "illegal receive port configured"); logmsg(LOG_ERR, "illegal receive port configured");
exit(-4); return -2;
} }
memset(&servaddr, 0, sizeof(servaddr)); memset(&servaddr, 0, sizeof(servaddr));
@ -60,16 +79,16 @@ void initReceiver() {
if (-1 == bind(receiveSockFd, (const struct sockaddr *) &servaddr, sizeof(servaddr))) { if (-1 == bind(receiveSockFd, (const struct sockaddr *) &servaddr, sizeof(servaddr))) {
logmsg(LOG_ERR, "unable to bind receive: %d", errno); logmsg(LOG_ERR, "unable to bind receive: %d", errno);
exit(-5); return -3;
} }
} }
int receiveAndVerifyMinuteBuffer(t_minuteBuffer *buf) { int receiveAndVerifyMinuteBuffer(t_receiverHandle *handle, t_minuteBuffer *buf) {
struct sockaddr_in cliaddr; struct sockaddr_in cliaddr;
socklen_t cliaddrlen = sizeof(cliaddr); socklen_t cliaddrlen = sizeof(cliaddr);
int n = recvfrom(receiveSockFd, buf->b, sizeof(buf->b), MSG_TRUNC, int n = recvfrom(handle->receiveSockFd, buf->b, sizeof(buf->b), MSG_TRUNC,
(struct sockaddr *) &cliaddr, &cliaddrlen); (struct sockaddr *) &cliaddr, &cliaddrlen);
logmsg(LOG_INFO, "received %d octets from %d.%d.%d.%d", logmsg(LOG_INFO, "received %d octets from %d.%d.%d.%d",
n, n,
(cliaddr.sin_addr.s_addr & 0x0ff), (cliaddr.sin_addr.s_addr & 0x0ff),
@ -82,7 +101,7 @@ int receiveAndVerifyMinuteBuffer(t_minuteBuffer *buf) {
return -1; return -1;
} }
config_setting_t *deviceConfig = config_setting_get_member(devicesConfig, buf->s.deviceId); config_setting_t *deviceConfig = config_setting_get_member(handle->devicesConfig, buf->s.deviceId);
if (deviceConfig == NULL) { if (deviceConfig == NULL) {
logmsg(LOG_INFO, "Unknown device: %s", buf->s.deviceId); logmsg(LOG_INFO, "Unknown device: %s", buf->s.deviceId);
return -2; return -2;
@ -93,7 +112,7 @@ int receiveAndVerifyMinuteBuffer(t_minuteBuffer *buf) {
logmsg(LOG_ERR, "No sharedsecret configured for device %s", buf->s.deviceId); logmsg(LOG_ERR, "No sharedsecret configured for device %s", buf->s.deviceId);
return -3; return -3;
} }
logmsg(LOG_INFO, "SharedSecret is %s", sharedSecret); // logmsg(LOG_INFO, "SharedSecret is %s", sharedSecret);
if (strlen(sharedSecret) >= SHA256_BLOCK_SIZE) { if (strlen(sharedSecret) >= SHA256_BLOCK_SIZE) {
logmsg(LOG_ERR, "Configured sharedsecret for device %s is too long", buf->s.deviceId); logmsg(LOG_ERR, "Configured sharedsecret for device %s is too long", buf->s.deviceId);
@ -118,7 +137,31 @@ int receiveAndVerifyMinuteBuffer(t_minuteBuffer *buf) {
return 0; return 0;
} }
int forwardMinuteBuffer(t_minuteBuffer *buf) { int initForwarder(config_t *cfg, t_forwarderHandle *handle) {
config_lookup_string(cfg, "influxUser", handle->influxUser);
config_lookup_string(cfg, "influxPass", handle->influxPass);
config_lookup_string(cfg, "influxServer", handle->influxServer);
config_lookup_int(cfg, "influxPort", &(handle->influxPort));
config_lookup_string(cfg, "influxDatabase", handle->influxDatabase);
config_lookup_string(cfg, "influxMeasurement", handle->influxMeasurement);
if (! handle->influxServer) {
logmsg("no influxServer configured");
return -1;
}
if (! handle->influxDatabase) {
logmsg("no influxDatabase configured");
return -2;
}
if (! handle->influxMeasurement) {
logmsg("no influxMeasurement configured");
return -3;
}
return 0;
}
int forwardMinuteBuffer(t_forwarderHandle *handle, t_minuteBuffer *buf) {
logmsg(LOG_INFO, "DeviceId: %s", buf->s.deviceId); logmsg(LOG_INFO, "DeviceId: %s", buf->s.deviceId);
logmsg(LOG_INFO, "Location: %s", buf->s.location); logmsg(LOG_INFO, "Location: %s", buf->s.location);
for (uint8_t j = 0; j < SECONDS_PER_MINUTE; j++) { for (uint8_t j = 0; j < SECONDS_PER_MINUTE; j++) {
@ -128,17 +171,28 @@ int forwardMinuteBuffer(t_minuteBuffer *buf) {
return 0; return 0;
} }
int main() { int main() {
readConfig(); if (0 != readConfig(&cfg)) {
logmsg(LOG_ERR, "error when reading configuration");
exit(-1);
}
initReceiver(); if (0 != initReceiver(&cfg, &receiveHandle)) {
logmsg(LOG_ERR, "error when initializing receiver");
exit(-2);
}
if (0 != initForwarder(&cfg, &forwarderHandle)) {
logmsg(LOG_ERR, "error when initializing forwarder");
exit(-2);
}
while (1) { while (1) {
t_minuteBuffer buf; t_minuteBuffer buf;
if (receiveAndVerifyMinuteBuffer(&buf) < 0) { if (receiveAndVerifyMinuteBuffer(&receiveHandle &buf) < 0) {
logmsg(LOG_ERR, "error in receiveAndVerify"); logmsg(LOG_ERR, "error in receiveAndVerify");
} else { } else {
if (forwardMinuteBuffer(&buf) < 0) { if (forwardMinuteBuffer(&forwarderHandle, &buf) < 0) {
logmsg(LOG_ERR, "error in send"); logmsg(LOG_ERR, "error in send");
} }
} }

View File

@ -1,5 +1,5 @@
influxUser = ""; // influxUser = "";
influxPass = ""; // influxPass = "";
influxServer = "127.0.0.1"; influxServer = "127.0.0.1";
influxPort = 8086; influxPort = 8086;
influxDatabase = "smarthome2"; influxDatabase = "smarthome2";