378 lines
8.6 KiB
C
378 lines
8.6 KiB
C
|
|
#include <stdio.h>
|
|
#include <stdlib.h>
|
|
#include <string.h>
|
|
#include <unistd.h>
|
|
#include <time.h>
|
|
#include <sys/socket.h>
|
|
#include <sys/un.h>
|
|
#include <poll.h>
|
|
#include <errno.h>
|
|
|
|
#include "dataserver_message.h"
|
|
#include "dataserver_client.h"
|
|
|
|
#define ds_inf(_fmt_,...) printf("I %s:%d "_fmt_"\n", __func__, __LINE__, ##__VA_ARGS__)
|
|
#define ds_err(_fmt_,...) printf("E %s:%d "_fmt_"\n", __func__, __LINE__, ##__VA_ARGS__)
|
|
|
|
#define ds_conn_inf(_fmt_,...) printf("I %s:%d conn/fd[%p/%d] "_fmt_"\n", __func__, __LINE__, conn, conn? conn->fd:-1, ##__VA_ARGS__)
|
|
#define ds_conn_err(_fmt_,...) printf("E %s:%d conn/fd[%p/%d] "_fmt_"\n", __func__, __LINE__, conn, conn? conn->fd:-1, ##__VA_ARGS__)
|
|
|
|
typedef struct {
|
|
int magic;
|
|
#define DS_CONN_MAGIC ((int)('d'<<24|'a'<<16)|'s'<<8|'r')
|
|
|
|
int fd;
|
|
int evt_fd;
|
|
} ds_connection_t;
|
|
|
|
static char *ds_server_path_default = "/run/ccdataserver.sock";
|
|
|
|
/*ret:
|
|
* -1:bad param,
|
|
* -2:timeout,
|
|
* -3:need re-connect
|
|
*/
|
|
static int dataserver_read(ds_connection_t *conn,
|
|
uint8_t *buf, size_t size, int timeout_ms)
|
|
{
|
|
int left = size;
|
|
int fd = conn->fd;
|
|
int evt_fd = conn->evt_fd;
|
|
|
|
for (;;) {
|
|
struct pollfd fds[2];
|
|
|
|
memset(fds, 0, sizeof(fds));
|
|
fds[0].fd = fd;
|
|
fds[0].events = POLLIN | POLLERR;
|
|
fds[1].fd = evt_fd;
|
|
fds[1].events = POLLIN | POLLERR;
|
|
|
|
int ret = poll(fds, 2, timeout_ms);
|
|
|
|
if (ret == 0)
|
|
return -2;
|
|
else if (ret < 0)
|
|
return -3;
|
|
|
|
ret = 0;
|
|
|
|
if (!(fds[0].revents & POLLIN))
|
|
return -3;
|
|
|
|
{
|
|
ret = read(fd, buf + (size - left), left);
|
|
if (ret <= 0) {
|
|
if (errno == EAGAIN)
|
|
continue;
|
|
return -3;
|
|
} else if (ret < left) {
|
|
left -= ret;
|
|
} else if (ret == left) {
|
|
return size;
|
|
}
|
|
}
|
|
}
|
|
return size;
|
|
}
|
|
|
|
/*ret:
|
|
* -1:bad param,
|
|
* -2:timeout,
|
|
* -3:need re-connect
|
|
*/
|
|
static int dataserver_write(ds_connection_t *conn, uint8_t *buf, size_t size)
|
|
{
|
|
struct pollfd fds[1];
|
|
fds[0].fd = conn->fd;
|
|
fds[0].events = POLLOUT | POLLERR;
|
|
|
|
int ret = poll(fds, 1, 1000);
|
|
|
|
if (ret == 0)
|
|
return -2;
|
|
else if (ret < 0)
|
|
return -3;
|
|
|
|
ret = 0;
|
|
|
|
if (fds[0].revents & POLLOUT) {
|
|
int sent = 0;
|
|
while (sent < size) {
|
|
ret = write(conn->fd, buf + sent, size - sent);
|
|
if (ret <= 0) {
|
|
if (errno == EAGAIN)
|
|
continue;
|
|
return -3;
|
|
} else {
|
|
sent += ret;
|
|
}
|
|
}
|
|
}
|
|
|
|
return size;
|
|
}
|
|
|
|
|
|
static uint32_t ds_time_now(void)
|
|
{
|
|
struct timespec ts;
|
|
clock_gettime(CLOCK_MONOTONIC, &ts);
|
|
return ts.tv_sec * 1000 + ts.tv_nsec / 1000000;
|
|
}
|
|
|
|
ds_conn_t* dataserver_connect(void)
|
|
{
|
|
ds_inf("connecting...");
|
|
|
|
char *server_path = getenv("CCDATASERVER_SOCKET");
|
|
if (!server_path) {
|
|
server_path = ds_server_path_default;
|
|
ds_inf("env CCDATASERVER_SOCKET not found, use default:\"%s\"", server_path);
|
|
}
|
|
|
|
int sockfd = -1;
|
|
if ((sockfd = socket(AF_UNIX, SOCK_STREAM, 0)) < 0) {
|
|
ds_err("open socket fail");
|
|
return NULL;
|
|
}
|
|
|
|
struct sockaddr_un s_un;
|
|
int len;
|
|
|
|
memset(&s_un, 0, sizeof(s_un));
|
|
s_un.sun_family = AF_UNIX;
|
|
strncpy(s_un.sun_path, server_path, sizeof(s_un.sun_path) - 1);
|
|
len = offsetof(struct sockaddr_un, sun_path) + strlen(s_un.sun_path);
|
|
|
|
uint32_t start = ds_time_now();
|
|
|
|
while (1) {
|
|
if (connect(sockfd, (struct sockaddr *)&s_un, len) == 0)
|
|
break;
|
|
|
|
ds_err("connect fail, retry...");
|
|
|
|
uint32_t now = ds_time_now();
|
|
|
|
if (now - start > 1000) {
|
|
close(sockfd);
|
|
return NULL;
|
|
}
|
|
|
|
usleep(100 * 1000);
|
|
}
|
|
|
|
ds_connection_t *conn = calloc(1, sizeof(ds_connection_t));
|
|
if (!conn) {
|
|
ds_err("mem fail");
|
|
close(sockfd);
|
|
return NULL;
|
|
}
|
|
|
|
conn->magic = DS_CONN_MAGIC;
|
|
conn->fd = sockfd;
|
|
conn->evt_fd = eventfd(0, 0);
|
|
|
|
ds_conn_inf("connected.");
|
|
|
|
return (ds_conn_t *)conn;
|
|
}
|
|
|
|
/*ret:
|
|
* -1:bad param,
|
|
* -2:timeout,
|
|
* -3:need re-connect
|
|
*/
|
|
int dataserver_read_message(ds_conn_t *connection, void *buf, size_t size, int timeout_ms)
|
|
{
|
|
ds_connection_t *conn = (ds_connection_t *)connection;
|
|
|
|
if (!VALID_DS_CONN(conn) || conn->magic != DS_CONN_MAGIC || (!buf && size) ) {
|
|
ds_err("bad parameter");
|
|
return -1;
|
|
}
|
|
|
|
if (!size)
|
|
return 0;
|
|
|
|
int count = 0;
|
|
|
|
ds_msg_head_t msg;
|
|
|
|
int ret = dataserver_read(conn, (uint8_t *)&msg, sizeof(ds_msg_head_t), timeout_ms);
|
|
|
|
if (ret <= -3) {
|
|
ds_conn_err("read msg head fail (%d)", ret);
|
|
return ret;
|
|
} else if (ret == -2)
|
|
/*time out*/
|
|
return ret;
|
|
else if (ret <= 0)
|
|
return ret;
|
|
|
|
if (ret == sizeof(ds_msg_head_t)) {
|
|
/*fill the header*/
|
|
memcpy(buf, &msg, ret);
|
|
count += ret;
|
|
|
|
if (msg.size) {
|
|
if ((msg.size + sizeof(ds_msg_head_t)) > size) {
|
|
ds_conn_err("insufficient buffer, required[%zd] now[%zd]", msg.size + sizeof(ds_msg_head_t), size);
|
|
return -3;
|
|
}
|
|
|
|
/*fill the body*/
|
|
ret = dataserver_read(conn, (uint8_t *)buf + sizeof(ds_msg_head_t), msg.size, timeout_ms);
|
|
if (ret == msg.size)
|
|
count += ret;
|
|
else {
|
|
ds_conn_err("read msg body fail (size:%d), err(%d)", msg.size, ret);
|
|
}
|
|
}
|
|
}
|
|
|
|
return (ret < 0) ? ret : count;
|
|
}
|
|
|
|
/*ret:
|
|
* -1:bad param,
|
|
* -2:timeout,
|
|
* -3:need re-connect
|
|
*/
|
|
int dataserver_write_message(ds_conn_t *connection, void *buf, size_t size)
|
|
{
|
|
ds_connection_t *conn = (ds_connection_t *)connection;
|
|
|
|
if (!VALID_DS_CONN(conn) || conn->magic != DS_CONN_MAGIC || (!buf && size != 0)) {
|
|
ds_err("bad parameter");
|
|
return -1;
|
|
}
|
|
|
|
if (size == 0)
|
|
return 0;
|
|
|
|
int ret = dataserver_write(conn, (uint8_t *)buf, size);
|
|
|
|
return ret;
|
|
}
|
|
|
|
int dataserver_disconnect(ds_conn_t *connection)
|
|
{
|
|
ds_connection_t *conn = (ds_connection_t *)connection;
|
|
|
|
ds_conn_inf("disconnecting...");
|
|
|
|
if (!VALID_DS_CONN(conn) || conn->magic != DS_CONN_MAGIC) {
|
|
ds_err("bad parameter");
|
|
return -1;
|
|
}
|
|
|
|
if (conn->fd >= 0) {
|
|
int64_t pad = 1;
|
|
write(conn->evt_fd, &pad, sizeof(pad));
|
|
|
|
close(conn->fd);
|
|
conn->fd = -1;
|
|
|
|
if (conn->evt_fd >= 0) {
|
|
close(conn->evt_fd);
|
|
conn->evt_fd = -1;
|
|
}
|
|
|
|
conn->magic = 0;
|
|
free(conn);
|
|
conn = NULL;
|
|
}
|
|
|
|
ds_conn_inf("disconnected.");
|
|
|
|
return 0;
|
|
}
|
|
|
|
int dataserver_destroy(ds_conn_t *connection)
|
|
{
|
|
ds_connection_t *conn = (ds_connection_t *)connection;
|
|
|
|
ds_conn_inf("destroying...");
|
|
/*
|
|
if (!VALID_DS_CONN(conn) || conn->magic != DS_CONN_MAGIC) {
|
|
ds_err("bad parameter");
|
|
return -1;
|
|
}
|
|
*/
|
|
|
|
ds_inf("destroyed.");
|
|
|
|
return 0;
|
|
}
|
|
|
|
#include <arpa/inet.h>
|
|
|
|
ds_conn_t* dataserver_connect_ip(uint32_t addr, uint32_t port)
|
|
{
|
|
ds_inf("connecting...");
|
|
|
|
int sockfd = -1;
|
|
if ((sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
|
|
ds_err("open socket fail");
|
|
return NULL;
|
|
}
|
|
|
|
struct sockaddr_in s_in;
|
|
|
|
memset(&s_in, 0, sizeof(s_in));
|
|
s_in.sin_family = AF_INET;
|
|
s_in.sin_port = htons(port);
|
|
s_in.sin_addr.s_addr = addr;
|
|
|
|
uint32_t start = ds_time_now();
|
|
|
|
while (1) {
|
|
if (connect(sockfd, (struct sockaddr *)&s_in, sizeof(s_in)) == 0)
|
|
break;
|
|
|
|
ds_err("connect fail, retry...");
|
|
|
|
uint32_t now = ds_time_now();
|
|
|
|
if (now - start > 1000) {
|
|
close(sockfd);
|
|
return NULL;
|
|
}
|
|
|
|
usleep(100 * 1000);
|
|
}
|
|
|
|
#ifdef BUILD_FOR_ANDROID
|
|
{
|
|
int val = 1;
|
|
errno = 0;
|
|
if (setsockopt(sockfd, SOL_SOCKET, SO_KEEPALIVE, &val, sizeof(val))) {
|
|
ds_err("setsockopt SO_KEEPALIVE failed. err(%d/%s)", errno, strerror(errno));
|
|
}
|
|
ds_err("setsockopt SO_KEEPALIVE failed. err(%d/%s)", errno, strerror(errno));
|
|
int keepidle = 20;
|
|
int keepinterval = 3;
|
|
setsockopt(sockfd, SOL_TCP, TCP_KEEPIDLE, &keepidle, sizeof(keepidle));
|
|
setsockopt(sockfd, SOL_TCP, TCP_KEEPINTVL, &keepinterval, sizeof(keepinterval));
|
|
}
|
|
#endif
|
|
|
|
ds_connection_t *conn = calloc(1, sizeof(ds_connection_t));
|
|
if (!conn) {
|
|
ds_err("mem fail");
|
|
close(sockfd);
|
|
return NULL;
|
|
}
|
|
|
|
conn->magic = DS_CONN_MAGIC;
|
|
conn->fd = sockfd;
|
|
conn->evt_fd = eventfd(0, 0);
|
|
|
|
ds_conn_inf("connected.");
|
|
|
|
return (ds_conn_t *)conn;
|
|
}
|