update

wangzhengquan

2020-07-17 1d1c02c006dd9561f8d50ffda5b16e29d81997fd

update

 queue/include/mod_socket.h

@@ -30,7 +30,7 @@

int mod_connect(void * _socket, int port);

int mod_send(void * _socket, void *buf, int size);
int mod_send(void * _socket, const void *buf, const int size);

int mod_recv(void * _socket, void **buf, int *size) ;


 queue/include/shm_socket.h

@@ -84,7 +84,7 @@

int shm_connect(shm_socket_t * socket, int port);

int shm_send(shm_socket_t * socket, void *buf, int size) ;
int shm_send(shm_socket_t * socket, const void *buf, const int size) ;

int shm_recv(shm_socket_t * socket, void **buf, int *size) ;


 queue/libshm_queue.a

Binary files differ

 queue/mod_socket.c

@@ -110,7 +110,7 @@

}

int mod_send(void * _socket, void *buf, int size) {
int mod_send(void * _socket, const void *buf, const int size) {
    mod_socket_t * socket = (mod_socket_t *) _socket;
    std::map<int, shm_socket_t* > *clientSocketMap = socket->shm_socket->clientSocketMap;
    std::map<int, shm_socket_t* >::iterator iter;
@@ -122,13 +122,14 @@
                rv = shm_send(socket->client_socket, buf, size);
                SemUtil::inc(socket->slots);
                break;
            case SURVEY:
            case PUB_SUB:
                for(iter = clientSocketMap->begin(); iter != clientSocketMap->end(); iter++) {
                    rv = shm_send(iter->second, buf, size);
                }
                break;
            default:
                err_exit(0, "不支持的模式%d", socket->mod);
                rv = shm_send(socket->client_socket, buf, size);
        }
        return rv;
        
@@ -158,8 +159,12 @@
            case PUB_SUB:
                rv = 0;
                break;

            case SURVEY:
            default:
                err_exit(0, "不支持的模式%d", socket->mod);
                rv = socket->recvQueue->pop(entry);
                *buf = entry.buf;
                *size = entry.size;
        }

        return rv;

 queue/shm_socket.c

@@ -306,7 +306,7 @@

 

int shm_send(shm_socket_t *socket, void *buf, int size) {
int shm_send(shm_socket_t *socket, const void *buf, const int size) {
    // hashtable_t *hashtable = mm_get_hashtable();
    if(socket->remoteQueue == NULL) {
        err_msg(errno, "当前客户端无连接!");

 test2/pub_sub

Binary files differ

 test2/pub_sub.c

@@ -50,6 +50,7 @@

  if (strcmp("client", argv[1]) == 0)
     client(port);
   
 shm_destroy();
 // fprintf(stderr, "Usage: reqrep %s|%s <URL> ...\n", "server", "client");
  return 0;

 test2/req_rep

Binary files differ

 test2/sub

Binary files differ

 test2/sub.c

File was deleted

 test2/survey.c

New file
@@ -0,0 +1,86 @@
#include "mod_socket.h"
#include "shm_mm.h"
#include "usg_common.h"

#define SERVER "server"
#define CLIENT "client"
#define DATE "DATE"

char *date(void) {
  time_t now = time(&now);
  struct tm *info = localtime(&now);
  char *text = asctime(info);
  text[strlen(text) - 1] = '\0'; // remove '\n'
  return (text);
}

int server(const int port) {
  int rv;

  void *socket = mod_open_socket(PUB_SUB);
  mod_socket_bind(socket, port);
  if ((rv = mod_listen(socket)) != 0) {
    printf("mod_listen");
  }
  for (;;) {
    printf("SERVER: SENDING DATE SURVEY REQUEST\n");
    if ((rv = mod_send(socket, DATE, strlen(DATE) + 1)) != 0) {
      printf("mod_send");
    }

    for (;;) {
      char *buf = NULL;
      int sz;
      rv = mod_recv(socket, (void **)&buf, &sz);
      
      if (rv != 0) {
        printf("mod_recv");
      }
      printf("SERVER: RECEIVED \"%s\" SURVEY RESPONSE\n", buf);
      mod_free(buf, sz);
    }

    printf("SERVER: SURVEY COMPLETE\n");
  }
}

int client(const int port, const char *name) {
  int rv;

  void *socket = mod_open_socket(PUB_SUB);
  if ((rv = mod_connect(socket, port)) != 0) {
    printf("mod_connect");
  }
  for (;;) {
    char *buf = NULL;
    int sz;
    if ((rv = mod_recv(socket, (void **)&buf, &sz)) == 0) {
      printf("CLIENT (%s): RECEIVED \"%s\" SURVEY REQUEST\n", name, buf);
      mod_free(buf, sz);
      char *d = date();
      printf("CLIENT (%s): SENDING DATE SURVEY RESPONSE\n", name);
      if ((rv = mod_send(socket, d, strlen(d) + 1, 0)) != 0) {
        printf("mod_send");
      }
    }
  }
}

int main(const int argc, const char **argv) {
  shm_init(512);
  int port;
  if ((argc >= 2) && (strcmp(SERVER, argv[1]) == 0)) {
    port = atoi(argv[2]);
    (server(port));
  }
  else if ((argc >= 3) && (strcmp(CLIENT, argv[1]) == 0)) {
    port = atoi(argv[2]);
    (client(port, argv[3]));
  } else {
    fprintf(stderr, "Usage: survey %s|%s <URL> <ARG> ...\n", SERVER, CLIENT);
  }

  shm_destroy();

  return 0;
}