update

wangzhengquan

2020-11-27 26927f7146adcf0170edf213cc77ad7368f5c076

update

 include/usgcommon/logger.h

@@ -17,12 +17,12 @@
  FILE *logFile;

  void dolog(const char *fmt, va_list ap, int level, int err = 0) {
    char buf[MAXBUF];
    char buf[MAXLINE];

    struct timeval tv;
    struct tm *info;
    gettimeofday(&tv, NULL);
    info = localtime(&(tv.tv_sec));
    info = localtime(&tv.tv_sec);
    strftime(buf, MAXBUF - 1, "%Y-%d-%m %H:%M:%S", info);
    snprintf(buf + strlen(buf), MAXBUF - strlen(buf) - 1, ",%ld [%s] ",  tv.tv_usec, strlevel(level));
    vsnprintf(buf + strlen(buf), MAXBUF - strlen(buf) - 1, fmt, ap);
@@ -42,6 +42,8 @@
   
    fflush(NULL); /* flushes all stdio output streams */
  }

  
  void init();

private:

 include/usgcommon/usg_common.h

@@ -98,8 +98,14 @@
char *rtrim(char *str, const char *seps);
char *trim(char *str, const char *seps);

/**
 * @ seperator 分割符
 */
char * str_join( const char *seperator, const char *first...);
char ** str_split(const char *str, const char *delim, int *arr_len);
/**
 * 把字符串以@delim分割为数组，数组的地址存放在@arr_addr
 */
int str_split(const char *str, const char *delim, char *** arr_addr);
char * array_join(char * const arr[], const char *seperator);
char * path_join(const char *path, ...);
// 反向查询

 lib/libusgcommon.a

Binary files differ

 libusgcommon.a

Binary files differ

 src/socket/dgram_mod_socket.h

@@ -115,7 +115,7 @@


/**
 * 获取soket端口号
 * 获取soket key
 */
int dgram_mod_get_port(void * _socket) ;


 src/socket/net_mod_socket.c

@@ -44,7 +44,7 @@
  net_mod_recv_msg_t ** recv_arr, int *recv_arr_size) {
  _sendandrecv_(node_arr,  arrlen, send_buf,send_size, recv_arr, recv_arr_size, -1);
}
int NetModSocket::sendandrecv_timout(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
int NetModSocket::sendandrecv_timeout(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
  net_mod_recv_msg_t ** recv_arr, int *recv_arr_size, int  timeout) {
  _sendandrecv_(node_arr,  arrlen, send_buf,send_size, recv_arr, recv_arr_size, timeout);
}

 src/socket/net_mod_socket.h

@@ -124,7 +124,7 @@
   * 如果建立连接的节点没有接受到消息等待timeout的时间后返回
   * @timeout 等待时间，单位是千分之一秒
  */
  int sendandrecv_timout(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
  int sendandrecv_timeout(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
    net_mod_recv_msg_t ** recv_arr, int *recv_arr_size, int  timeout);

  /**

 src/socket/net_mod_socket_wrapper.c

@@ -100,10 +100,10 @@
 * 如果建立连接的节点没有接受到消息等待timeout的时间后返回
 * @timeout 等待时间，单位是千分之一秒
*/
int net_mod_socket_sendandrecv_timout(void *_socket, net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
  net_mod_recv_msg_t ** recv_arr, int *recv_arr_size, int  timeout){
int net_mod_socket_sendandrecv_timeout(void *_socket, net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
  net_mod_recv_msg_t ** recv_arr, int *recv_arr_size,  int timeout){
    net_mod_socket_t *sockt = (net_mod_socket_t *)_socket;
    return sockt->sockt->sendandrecv_timout(node_arr,  arrlen, send_buf,  send_size, recv_arr, recv_arr_size, timeout);
    return sockt->sockt->sendandrecv_timeout(node_arr,  arrlen, send_buf,  send_size, recv_arr, recv_arr_size, timeout);
}

int net_mod_socket_sendandrecv_nowait(void *_socket, net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 

 src/socket/net_mod_socket_wrapper.h

@@ -74,8 +74,8 @@
 * 如果建立连接的节点没有接受到消息等待timeout的时间后返回
 * @timeout 等待时间，单位是千分之一秒
*/
int net_mod_socket_sendandrecv_timout(void *_sockt, net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
  net_mod_recv_msg_t ** recv_arr, int *recv_arr_size, int  timeout);
int net_mod_socket_sendandrecv_timeout(void *_sockt, net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
  net_mod_recv_msg_t ** recv_arr, int *recv_arr_size, int timeout);

/**
 * 不等待立即返回
@@ -93,7 +93,6 @@
 * @return 成功发布的节点的个数
 */
int net_mod_socket_pub(void *_sockt, net_node_t *node_arr, int node_arr_len, char *topic, int topic_size, void *content, int content_size);



/**

 test_net_socket/net_mod_socket.sh

@@ -1,21 +1,33 @@
function server() {
	 

# 打开请求应答的接受端
    ./test_net_mod_socket --fun="start_reply" --key=11 & server_pid=$! &&  echo "pid: ${server_pid}" 
	
    
# 开启bus 
 ./test_net_mod_socket --fun="start_bus_server" --key=8  & server_pid=$! &&  echo "pid: ${server_pid}"
	 
	 
# 开启网络转发代理
    ./test_net_mod_socket  --fun="start_net_proxy" --port=5000 & server_pid=$! && echo "pid: ${server_pid}" 
	 


# 打开请求应答测试的接受端
    ./test_net_mod_socket --fun="start_reply" --key=11 & server_pid=$! &&  echo "pid: ${server_pid}" 
}

function client() {
    ./test_net_mod_socket client 5000

    # ./test_net_mod_socket --fun="start_net_client" \
    #  --sendlist="192.168.5.10:5000:11, 192.168.5.22:5000:11, 192.168.20.104:5000:11" \
    #  --publist="192.168.5.10:5000:8, 192.168.5.22:5000:8, 192.168.20.104:5000:8"


    ./test_net_mod_socket --fun="start_net_client" \
     --sendlist="localhost:5000:11" \
     --publist="localhost:5000:8"  

	 
}

function mclient() {
    ./test_net_mod_socket --fun="start_net_mclient" \
     --sendlist="localhost:5000:11"
	 
}

function close() {
@@ -37,6 +49,11 @@
  "client")
     client
  ;;
  "mclient")
    close
    server 
    mclient
  ;;
  "close")
     close
  ;;
@@ -45,6 +62,7 @@
    server 
    client
  ;;

  *)
  echo "error input"
  exit 1

 test_net_socket/test_net_mod_socket.c

@@ -5,10 +5,25 @@
#include <getopt.h>

typedef struct Targ {
    int port;
    char *sendlist;
    int id;

}Targ;

struct argument_t {
  char *fun;
  int port;
  int key;
  char *sendlist;
  char *publist;
  char **cmd_arr;
  int cmd_arr_len;
};

argument_t parse_args (int argc, char *argv[]);
void usage(char *name);
int parse_node_list(char *str, net_node_t *node_arr_addr[]) ;
void print_node_list(net_node_t *node_arr, int len);

void start_net_proxy(int port) {
  printf("Start net proxy\n");
@@ -18,7 +33,20 @@
    }
}

void start_net_client(int port ){

void *print_sub_msg(void *sockt) {
  pthread_detach(pthread_self());
  void *recvbuf;
  int size;
  int key;
  while (net_mod_socket_recvfrom( sockt, &recvbuf, &size, &key) == 0) {
    printf("收到订阅消息:%s\n", recvbuf);
    free(recvbuf);
  }
  
}

void start_net_client(char *sendlist, char*publist ){
    void * client = net_mod_socket_open();
    char content[MAXLINE];
    char action[512];
@@ -26,27 +54,24 @@
  int buskey;
    
    int recv_arr_size, i, n;


    net_mod_recv_msg_t *recv_arr;
    //192.168.20.104
  int node_arr_size = 3;
    net_node_t node_arr[] = {
        {"192.168.5.22", port, 11},
        {"192.168.20.10", port, 11},
        {"192.168.20.104", port, 11}
    };

    int pub_node_arr_size = 3;
    net_node_t pub_node_arr[] = {
        {"192.168.5.22", port, 8},
        {"192.168.20.10", port, 8},
        {"192.168.20.104", port, 8}
    };
  pthread_t tid;
  pthread_create(&tid, NULL, print_sub_msg, client);
	 
  //192.168.5.10:5000:11, 192.168.5.22:5000:11, 192.168.5.104:5000:11
  net_node_t *node_arr;
  int node_arr_size = parse_node_list(sendlist, &node_arr);
    // print_node_list(node_arr, node_arr_size);

  //192.168.5.10:5000:8, 192.168.5.22:5000:8, 192.168.5.104:5000:8
  net_node_t *pub_node_arr;
    int pub_node_arr_size = parse_node_list(publist, &pub_node_arr);
  // print_node_list(pub_node_arr, pub_node_arr_size);
    
  while (true) {
    //printf("Usage: pub <topic> [content] or sub <topic>\n");
    printf("Can I help you? pub, send or quit\n");
    printf("Can I help you? pub,sub,desub,send or quit\n");
    scanf("%s",action);
    
    if(strcmp(action, "pub") == 0) {
@@ -62,7 +87,8 @@
        
          if (fgets(content, MAXLINE, stdin) != NULL) {
              // 收到消息的节点即使没有对应的信息， 也要回复一个表示无的消息,否则会一直等待
            n = net_mod_socket_sendandrecv(client, node_arr, node_arr_size, content, strlen(content), &recv_arr, &recv_arr_size);
            n = net_mod_socket_sendandrecv_timeout(client, node_arr, node_arr_size, content, 
          strlen(content), &recv_arr, &recv_arr_size, 5000);
            printf("send %d nodes\n", n);
            for(i=0; i<recv_arr_size; i++) {
                printf("host:%s, port: %d, key:%d, content: %s\n", 
@@ -78,9 +104,9 @@
          }
    }
    else if(strcmp(action, "desub") == 0) {
      printf("Please input buskey topic!\n");
      printf("Please input buskey and topic!\n");
       
      scanf("%d %s", buskey, topic);
      scanf("%d %s", &buskey, topic);
      if (net_mod_socket_desub(client, topic, strlen(topic),  buskey) == 0) {
         printf("%d Desub success!\n", net_mod_socket_get_key(client));
      } else {
@@ -90,8 +116,10 @@
     
    } 
    else if(strcmp(action, "sub") == 0) {
      printf("Please input topic!\n");
      scanf("%s", topic);
      printf("Please input buskey and topic!\n");
      scanf("%d %s",&buskey, topic);

      printf("===%d %s\n",buskey, topic);
      if (net_mod_socket_sub(client, topic, strlen(topic),  buskey) == 0) {
         printf("%d Sub success!\n", net_mod_socket_get_key(client));
      } else {
@@ -118,19 +146,15 @@

void *runclient(void *arg) {
  Targ *targ = (Targ *)arg;
  int port = targ->port;
  char sendbuf[512];
 
  int i,j, n, recv_arr_size;
  net_mod_recv_msg_t *recv_arr;

  int node_arr_size = 1;
    //192.168.20.104
    net_node_t node_arr[] = {
        //{0, port, 11}
    {"192.168.20.10", port, 11}
    // {"192.168.20.104", port, 11}
    };
 

  net_node_t *node_arr;
  int node_arr_size = parse_node_list(targ->sendlist, &node_arr);

  void * client = net_mod_socket_open();
    
@@ -163,9 +187,9 @@
  return (void *)i;
}

void start_net_mclient(int port) {
void start_net_mclient(char *sendlist) {

  int status, i = 0, processors = 1;
  int status, i = 0, processors = 4;
  void *res[processors];
  // Targ *targs = (Targ *)calloc(processors, sizeof(Targ));
  Targ targs[processors];
@@ -177,7 +201,7 @@
printf("开始测试...\n");  
  gettimeofday(&start, NULL);
  for (i = 0; i < processors; i++) {
    targs[i].port = port;
    targs[i].sendlist = sendlist;
    targs[i].id = i;
    pthread_create(&tids[i], NULL, runclient, (void *)&targs[i]);
  }
@@ -212,6 +236,7 @@
 

void start_reply(int key) {
  printf("start reply\n");
  void *socket = net_mod_socket_open();
  net_mod_socket_bind(socket, key);
  int size;
@@ -229,121 +254,7 @@



void usage(char *name)
{
  fprintf(stderr, "Usage: %s  [OPTIONS]  [ARG...]\n\n", name);
  fprintf(stderr, "Test net mod socket\n\n");
  fprintf(stderr, "Options:\n\n");
  #define fpe(str) fprintf(stderr, "  %s", str);
  fpe("-f, --funciton                       Function name\n");
  fpe("-p, --port                           TCP/IP Port\n");
  fpe("-k, --key                            SHM Key\n");
  fpe("\n");
}

struct argument_t {
  char *fun;
  int port;
  int key;
  char **cmd_arr;
  int cmd_arr_len;
};

argument_t parse_args (int argc, char *argv[])
{
  int c;

  if(argc < 2) {
    usage(argv[0]);
    exit(1);
  }

  if(argc == 2 && strcmp(argv[1], "--help") == 0) {
    usage(argv[0]);
    exit(0);
  }

  
  argument_t mopt = {};
  
  // mopt.volume_list_size = 0;

  opterr = 0;

  static struct option long_options[] =
  {
    /* These options set a flag. */
     
    {"fun",  required_argument, 0, 'f'},
    {"key",  required_argument, 0, 'k'},
    {"port",  required_argument, 0, 'p'},
    {0, 0, 0, 0}
  };
  /* getopt_long stores the option index here. */
  int option_index = 0;
  while (1)
  {
    

    c = getopt_long (argc, argv, "+f:k:p:", long_options, &option_index);

    /* Detect the end of the options. */
    if (c == -1)
      break;

    switch (c)
    {
    case 0:
      // printf("ffffffff\n");
      /* If this option set a flag, do nothing else now. */
      if (long_options[option_index].flag != 0)
        break;
      printf ("option %s", long_options[option_index].name);
      if (optarg)
        printf (" with arg %s", optarg);
      printf ("\n");
      break;
     
    case 'f':
      mopt.fun = optarg;
      break;

    case 'k':
      mopt.key = atoi(optarg);
      break;

    case 'p':
       // printf ("==name with value `%s'\n", optarg);
      mopt.port = atoi(optarg);
      break;

    case '?':
     // printf ("==? optopt=%c, %s, `%s', %d\n", optopt, optarg, argv[optind], optind);
      /* getopt_long already printed an error message. */
      usage(argv[0]);
      exit(1);
      break;

    default:
      //printf ("==default optopt=%c, %s, `%s'\n",optopt, optarg,  argv[optind]);
      break;
    }
  }

  // printf ("optind = %d, argc=%d \n", optind, argc);
  /* Print any remaining command line arguments (not options). */
  if (optind < argc)
  {
    mopt.cmd_arr = &argv[optind];
    mopt.cmd_arr_len = argc - optind;
    // printf ("non-option ARGV-elements: ");
    // while (optind < argc)
    //   printf ("%d, %d, %s \n", optind, argc, argv[optind++]);
    // putchar ('\n');
  }
  return mopt;

}
int main(int argc, char *argv[]) {
    shm_init(512);
 
@@ -374,13 +285,27 @@
    }
    start_reply(opt.key);
  }

  else if (strcmp("start_net_client", opt.fun) == 0) {
    if(opt.port == 0) {
    if(opt.sendlist == 0) {
      fprintf(stderr, "Missing sendlist .\n");
      usage(argv[0]);
      exit(1);
    }
    start_net_client(opt.port);
    if(opt.publist == 0) {
      fprintf(stderr, "Missing publist.\n");
      usage(argv[0]);
      exit(1);
    }
    start_net_client(opt.sendlist, opt.publist);
  }
  else if (strcmp("start_net_mclient", opt.fun) == 0) {
    if(opt.sendlist == 0) {
      fprintf(stderr, "Missing sendlist .\n");
      usage(argv[0]);
      exit(1);
    }
     
    start_net_mclient(opt.sendlist);
  }
  else {
    usage(argv[0]);
@@ -388,8 +313,166 @@

  }


}



void usage(char *name)
{
  fprintf(stderr, "Usage: %s  [OPTIONS]  [ARG...]\n\n", name);
  fprintf(stderr, "Test net mod socket\n\n");
  fprintf(stderr, "Options:\n\n");
  #define fpe(str) fprintf(stderr, "  %s", str);
  fpe("-f, --funciton                       Function name\n");
  fpe("-p, --port                           TCP/IP Port\n");
  fpe("-k, --key                            SHM Key\n");
  fpe("\n");
}



argument_t parse_args (int argc, char *argv[])
{
  int c;

  if(argc < 2) {
    usage(argv[0]);
    exit(1);
  }

  if(argc == 2 && strcmp(argv[1], "--help") == 0) {
    usage(argv[0]);
    exit(0);
  }

  
  argument_t mopt = {};
  
  // mopt.volume_list_size = 0;

  opterr = 0;


  static struct option long_options[] =
  {
    /* These options set a flag. */
     
    {"fun",  required_argument, 0, 'f'},
    {"key",  required_argument, 0, 'k'},
    {"port",  required_argument, 0, 'p'},
    {"sendlist",  required_argument, (int *)mopt.sendlist, 0},
    {"publist",  required_argument, (int *)mopt.publist, 0},
    {0, 0, 0, 0}
  };
  /* getopt_long stores the option index here. */
  int option_index = 0;
  while (1)
  {
    

    c = getopt_long (argc, argv, "+f:k:p:", long_options, &option_index);

    /* Detect the end of the options. */
    if (c == -1)
      break;

    switch (c)
    {
    case 0:
      /* If this option set a flag, do nothing else now. */
      if (long_options[option_index].flag != 0)
        break;
     
      if(strcmp(long_options[option_index].name, "sendlist") == 0) {
        mopt.sendlist = optarg;
      }
      else if(strcmp(long_options[option_index].name, "publist") == 0) {
        mopt.publist = optarg;
      }
      else {
        printf ("option %s", long_options[option_index].name);
        if (optarg)
          printf (" with arg %s", optarg);
        printf ("\n");
      }
    
      break;
     
    case 'f':
      mopt.fun = optarg;
      break;

    case 'k':
      mopt.key = atoi(optarg);
      break;

    case 'p':
       // printf ("==name with value `%s'\n", optarg);
      mopt.port = atoi(optarg);
      break;

    case '?':
     printf ("==? optopt=%c, %s, `%s', %d\n", optopt, optarg, argv[optind], optind);
      /* getopt_long already printed an error message. */
      usage(argv[0]);
      exit(1);
      break;

    default:
      //printf ("==default optopt=%c, %s, `%s'\n",optopt, optarg,  argv[optind]);
      break;
    }
  }

  // printf ("optind = %d, argc=%d \n", optind, argc);
  /* Print any remaining command line arguments (not options). */
  if (optind < argc)
  {
    mopt.cmd_arr = &argv[optind];
    mopt.cmd_arr_len = argc - optind;
    // printf ("non-option ARGV-elements: ");
    // while (optind < argc)
    //   printf ("%d, %d, %s \n", optind, argc, argv[optind++]);
    // putchar ('\n');
  }
  return mopt;

}


/**
 * @str "192.168.5.10:5000:11, 192.168.5.22:5000:11, 192.168.5.104:5000:11"
 * @node_arr_addr 返回处理后的网络节点数组 
 * {
 *   {"192.168.5.22", 5000, 11},
 *   {"192.168.20.10", 5000, 11},
 *   {"192.168.20.104", 5000, 11}
 * }
 * @return 数组的长度
 */
int parse_node_list(char *str, net_node_t *node_arr_addr[]) {
  int i, j;
  char **property_arr;
  int property_arr_len;
  char **entry_arr;
  int entry_arr_len = str_split(str, ",", &entry_arr);

  net_node_t *node_arr = (net_node_t *) calloc(entry_arr_len, sizeof(net_node_t));
  for(i = 0; i < entry_arr_len; i++) {
    property_arr_len = str_split(entry_arr[i], ":", &property_arr);
    node_arr[i] = {trim(property_arr[0], 0), atoi(property_arr[1]), atoi(property_arr[2])};
    free(entry_arr[i]);
    free(property_arr[1]);
    free(property_arr[2]);
  }
  *node_arr_addr = node_arr;


  return entry_arr_len;
}

void print_node_list(net_node_t *node_arr, int len) {
  for(int i = 0; i < len; i++) {
    printf("%s,%d,%d,\n", node_arr[i].host,  node_arr[i].port,  node_arr[i].key);
  }
}