update

wangzhengquan

2020-07-14 fb1289e6d069e1a0f14dd45ba07720f09e90ab56

update

 queue/Makefile

@@ -29,13 +29,14 @@

#static lib
$(LIBSQUEUE): $(OBJS)
    $(AR) rv $@ $? $(ROOT)/lib/libusgcommon.a
    $(AR) rv $@ $?
    $(RANLIB) $@

#dynamic lib
$(DLIBSQUEUE): $(SOURCES)
    rm -f *.o
    $(CC) -fPIC -shared $(CFLAGS) $^ -o $@ $(LDFLAGS) $(ROOT)/lib/libusgcommon.so
    $(CC) -fPIC -shared $(CFLAGS) $^ -o $@ $(LDFLAGS)
    #$(CC) -fPIC -shared $(CFLAGS) $(LDFLAGS)  -o $@ $^ -Wl,--whole-archive $(ROOT)/lib/libusgcommon.a -Wl,--no-whole-archive 


#PREFIX is environment variable, but if it is not set, then set default value

 queue/libshm_queue.a

Binary files differ

 queue/libshm_queue.so

Binary files differ

 test2/Makefile

@@ -2,11 +2,11 @@
# Makefile for common library.
#
ROOT=..
LDLIBS+=-Wl,-rpath=$(ROOT)/queue
LDLIBS+=-Wl,-rpath=$(ROOT)/queue:$(ROOT)/lib
# 开源工具包路径
LDDIR += -L$(ROOT)/queue
# 开源工具包
LDLIBS += -lshm_queue -lpthread
LDLIBS += -lshm_queue -lusgcommon -lpthread

INCLUDE += -I$(ROOT)/queue/ -I$(ROOT)/queue/include

@@ -14,7 +14,7 @@
include $(ROOT)/Make.defines.$(PLATFORM)


PROGS =    test_queue_wrapper server client
PROGS =    test_queue_wrapper server client pub sub


build: $(PROGS)

 test2/client

Binary files differ

 test2/client.c

@@ -1,6 +1,5 @@
#include "shm_queue_wrapper.h"
#include "mm.h"
 

typedef struct msg_t
{
    int key;

 test2/pub

Binary files differ

 test2/pub.c

New file
@@ -0,0 +1,62 @@
#include "shm_queue_wrapper.h"

 
typedef struct msg_t
{
    int key;
    char buf[100];
	
} msg_t;


static void * remote_queues[100];
static int remote_queues_len = 0;

void *pub(void *) {
    msg_t send_msg;
    send_msg.key = 1;
    while(true) {
        for(int i = 0; i < remote_queues_len; i++ ) {
            sprintf(send_msg.buf, "今日头条××××");
            shmqueue_push(remote_queues[i], (void *)&send_msg, sizeof(send_msg));
        }
        sleep(2);
    }
}

void server() {
     void * msg;
     int msg_size;
 	
     int key = 1;
    size_t qsize = 16;
    void * local_queue = shmqueue_create( &key, qsize);
	

    pthread_t tid;
     pthread_create(&tid, NULL, pub, NULL);
	
    while(shmqueue_pop(local_queue, &msg, &msg_size) ) {
        void * remote_queue = shmqueue_attach(((msg_t *)msg)->key);
        remote_queues[remote_queues_len++] = remote_queue;
        printf("收到订阅请求", ((msg_t *)msg)->buf);
	   
	   
        // shmqueue_drop(remote_queue);
       // cout <<  item.pic << endl;
        // i++;
    }

    //销毁队列
    shmqueue_drop(local_queue);
}

 
int main () {
    shm_init(512);
    server();

    //整个进程退出时需要执行这个方法，该方法首先会检查是否还有其他进程在使用该共享内存，如果还有其他进程在使用就只是detach,如果没有其他进程在使用则销毁整块内存。
    shm_destroy();
    return 0;
}

 test2/server

Binary files differ

 test2/server.c

@@ -1,5 +1,5 @@
#include "shm_queue_wrapper.h"
#include "mm.h"

 
typedef struct msg_t
{

 test2/sub

Binary files differ

 test2/sub.c

New file
@@ -0,0 +1,48 @@
#include "shm_queue_wrapper.h"

typedef struct msg_t
{
    int key;
    char buf[100];
	
} msg_t;

void client() {
    int key = -1;
    size_t qsize = 16;
    void * remote_queue = shmqueue_attach( 1);
    void * local_queue = shmqueue_create( &key, qsize);
    // message_t item;
    msg_t msg;
    msg.key = key;

    void * rec_msg;
    int rec_msg_size;

    sprintf(msg.buf, "我要订阅今日头条");
    shmqueue_push(remote_queue, (void *)&msg, sizeof(msg));
    //入队
    while(true) {
	  
      //printf("send: %s\n",  msg.buf);
      shmqueue_pop(local_queue, &rec_msg, &rec_msg_size);
      printf("收到订阅消息: %s\n",  ((msg_t*)rec_msg)->buf);
      free(rec_msg);

	  
    }
	 
    //销毁队列
    shmqueue_drop(local_queue);
    shmqueue_drop(remote_queue);
}

 
int main () {
    shm_init(512);
    client();

    //整个进程退出时需要执行这个方法，该方法首先会检查是否还有其他进程在使用该共享内存，如果还有其他进程在使用就只是detach,如果没有其他进程在使用则销毁整块内存。
    shm_destroy();
    return 0;
}

 test2/test_queue_wrapper

Binary files differ