complete

zhangmeng

2020-07-20 91a389294ddd1555349559d01141875bf2154c93

complete

 libcsoftbus.c

New file
@@ -0,0 +1,259 @@
#ifdef __cplusplus
extern "C"{
#endif

#include <dlfcn.h>
#include <stdlib.h>
#include <stdio.h>

#include "libcsoftbus.h"

#ifdef __cplusplus
}
#endif

#define check_only(fn, lib) \
do{ \
    if(!fn){ \
        dlclose(lib); \
        printf("run func %s error\n", #fn); \
        return; \
    } \
}while(0)

#define check_with_ret(fn, lib, flag) \
do{ \
    if(!fn){ \
        dlclose(lib); \
        printf("run func %s error\n", #fn); \
        return flag; \
    } \
}while(0)


hcsoftbus c_softbus_handle(const char *so){
    hcsoftbus lib = dlopen(so, RTLD_LAZY);
    if(lib){

        fn_shm_destroy = (tfn_shm_destroy)dlsym(lib, l_shm_destroy);
        check_with_ret(fn_shm_destroy, lib, NULL);
    }else{
        printf("softbus dlopen - %s\n", dlerror());
    }
    return lib;
}

void c_softbus_release(hcsoftbus lib){
    if(lib) {
        dlclose(lib);
    }
}

void wrap_fn_shm_init(hcsoftbus lib, int size){
    if (!fn_shm_init){
        fn_shm_init = (tfn_shm_init)dlsym(lib, l_shm_init);
        check_only(fn_shm_init, lib);
    }
    fn_shm_init(size);
}

void wrap_fn_shm_destroy(hcsoftbus lib){
    if (!fn_shm_destroy){
        printf("shm destroy failed\n");
        return;
    }
    fn_shm_destroy();
}

void wrap_fn_shm_rm_dead_queue(hcsoftbus lib, void *array, int len){
    if (!fn_shm_rm_dead_queue){
        fn_shm_rm_dead_queue = (tfn_shm_rm_dead_queue)dlsym(lib, l_shm_rm_dead_queue);
        check_only(fn_shm_rm_dead_queue, lib);
    }
    fn_shm_rm_dead_queue(array, len);
}

void* wrap_fn_queue_create(hcsoftbus lib, int *key, int queue_size){
    if (!fn_queue_create){
        fn_queue_create = (tfn_queue_create)dlsym(lib, l_queue_create);
        check_with_ret(fn_queue_create, lib, NULL);
    }
    return fn_queue_create(key, queue_size);
}

void* wrap_fn_queue_attach(hcsoftbus lib, int key){
    if (!fn_queue_attach){
        fn_queue_attach = (tfn_queue_attach)dlsym(lib, l_queue_attach);
        check_with_ret(fn_queue_attach, lib, NULL);
    }
    return fn_queue_attach(key);
}

void wrap_fn_queue_drop(hcsoftbus lib, void* shmq){
    if (!fn_queue_drop){
        fn_queue_drop = (tfn_queue_drop)dlsym(lib, l_queue_drop);
        check_only(fn_queue_drop, lib);
    }
    fn_queue_drop(shmq);
}

int wrap_fn_queue_size(hcsoftbus lib, void *shmq){
    if (!fn_queue_size){
        fn_queue_size = (tfn_queue_size)dlsym(lib, l_queue_size);
        check_with_ret(fn_queue_size, lib, 0);
    }
    return fn_queue_size(shmq);
}

int wrap_fn_queue_full(hcsoftbus lib, void *shmq){
    if (!fn_queue_full){
        fn_queue_full = (tfn_queue_full)dlsym(lib, l_queue_full);
        check_with_ret(fn_queue_full, lib, 1);
    }
    return fn_queue_full(shmq);
}

int wrap_fn_queue_empty(hcsoftbus lib, void *shmq){
    if (!fn_queue_empty){
        fn_queue_empty = (tfn_queue_empty)dlsym(lib, l_queue_empty);
        check_with_ret(fn_queue_empty, lib, 0);
    }
    return fn_queue_empty(shmq);
}

int wrap_fn_queue_push(hcsoftbus lib, void *shmq, void *ele, int ele_size){
    if (!fn_queue_push){
        fn_queue_push = (tfn_queue_push)dlsym(lib, l_queue_push);
        check_with_ret(fn_queue_push, lib, 0);
    }

    return fn_queue_push(shmq, ele, ele_size);
}

int wrap_fn_queue_push_nowait(hcsoftbus lib, void *shmq, void *ele, int ele_size){
    if (!fn_queue_push_nowait){
        fn_queue_push_nowait = (tfn_queue_push_nowait)dlsym(lib, l_queue_push_nowait);
        check_with_ret(fn_queue_push_nowait, lib, 0);
    }
    return fn_queue_push_nowait(shmq, ele, ele_size);
}

int wrap_fn_queue_push_timeout(hcsoftbus lib, void *shmq, void *ele, int ele_size, int sec, int nsec){
    if (!fn_queue_push_timeout){
        fn_queue_push_timeout = (tfn_queue_push_timeout)dlsym(lib, l_queue_push_timeout);
        check_with_ret(fn_queue_push_timeout, lib, 0);
    }
    return fn_queue_push_timeout(shmq, ele, ele_size, sec, nsec);
}

int wrap_fn_queue_pop(hcsoftbus lib, void *shmq, void **ele, int *ele_size){
    if (!fn_queue_pop){
        fn_queue_pop = (tfn_queue_pop)dlsym(lib, l_queue_pop);
        check_with_ret(fn_queue_pop, lib, 0);
    }
    return fn_queue_pop(shmq, ele, ele_size);
}

int wrap_fn_queue_pop_nowait(hcsoftbus lib, void *shmq, void **ele, int *ele_size){
    if (!fn_queue_pop_nowait){
        fn_queue_pop_nowait = (tfn_queue_pop_nowait)dlsym(lib, l_queue_pop_nowait);
        check_with_ret(fn_queue_pop_nowait, lib, 0);
    }
    return fn_queue_pop_nowait(shmq, ele, ele_size);
}

int wrap_fn_queue_pop_timeout(hcsoftbus lib, void *shmq, void **ele, int *ele_size, int sec, int nsec){
    if (!fn_queue_pop_timeout){
        fn_queue_pop_timeout = (tfn_queue_pop_timeout)dlsym(lib, l_queue_pop_timeout);
        check_with_ret(fn_queue_pop_timeout, lib, 0);
    }

    return fn_queue_pop_timeout(shmq, ele, ele_size, sec, nsec);
}

void wrap_fn_queue_free_poped(hcsoftbus lib, void *ptr){
    if (!fn_queue_free_poped){
        fn_queue_free_poped = (tfn_queue_free_poped)dlsym(lib, l_queue_free_poped);
        check_only(fn_queue_free_poped, lib);
    }
    fn_queue_free_poped(ptr);
}

////////////////////////////////////////////
// socket mode
////////////////////////////////////////////
void *wrap_fn_socket_open(hcsoftbus lib, int mod){
    if (!fn_socket_open) {
        fn_socket_open = (tfn_socket_open)dlsym(lib, l_socket_open);
        check_with_ret(fn_socket_open, lib, NULL);
    }
    return fn_socket_open(mod);
}

int wrap_fn_socket_close(hcsoftbus lib, void* s){
    if(!fn_socket_close){
        fn_socket_close = (tfn_socket_close)dlsym(lib, l_socket_close);
        check_with_ret(fn_socket_close, lib, -1);
    }
    return fn_socket_listen(s);
}

int wrap_fn_socket_bind(hcsoftbus lib, void *s, int port){
    if (!fn_socket_bind){
        fn_socket_bind = (tfn_socket_bind)dlsym(lib, l_socket_bind);
        check_with_ret(fn_socket_bind, lib, -1);
    }

    return fn_socket_bind(s, port);
}

int wrap_fn_socket_listen(hcsoftbus lib, void *s){
    if (!fn_socket_listen){
        fn_socket_listen = (tfn_socket_listen)dlsym(lib, l_socket_listen);
        check_with_ret(fn_socket_listen, lib, -1);
    }
    return fn_socket_listen(s);
}

int wrap_fn_socket_connect(hcsoftbus lib, void *s, int port){
    if (!fn_socket_connect){
        fn_socket_connect = (tfn_socket_connect)dlsym(lib, l_socket_connect);
        check_with_ret(fn_socket_connect, lib, -1);
    }
    return fn_socket_connect(s, port);
}

int wrap_fn_socket_send(hcsoftbus lib, void *s, const void *buf, const int size){
    if (!fn_socket_send){
        fn_socket_send = (tfn_socket_send)dlsym(lib, l_socket_send);
        check_with_ret(fn_socket_send, lib, -1);
    }

    return fn_socket_send(s, buf, size);
}

int wrap_fn_socket_recv(hcsoftbus lib, void *s, void **buf, int *size){
    if (!fn_socket_recv){
        fn_socket_recv = (tfn_socket_recv)dlsym(lib, l_socket_recv);
        check_with_ret(fn_socket_recv, lib, -1);
    }
    return fn_socket_recv(s, buf, size);
}

void wrap_fn_socket_buf_free(hcsoftbus lib, void *buf){
    if (!fn_socket_buf_free){
        fn_socket_buf_free = (tfn_socket_buf_free)dlsym(lib, l_socket_free);
    }
    if (fn_socket_buf_free){
        fn_socket_buf_free(buf);
    }
}

int wrap_fn_socket_port(hcsoftbus lib, void *s){
    if (!fn_socket_port){
        fn_socket_port = (tfn_socket_port)dlsym(lib, l_socket_port);
        check_with_ret(fn_socket_port, lib, -1);
    }

    return fn_socket_port(s);
}

 libcsoftbus.h

New file
@@ -0,0 +1,140 @@
#ifndef _c_libshm_queue_so_go_wrapper_h_
#define _c_libshm_queue_so_go_wrapper_h_

#include "libcsoftbus_func.h"
#include <stddef.h>

#ifdef __cplusplus
extern "C"{
#endif

// shm manipulate
static tfn_shm_init             fn_shm_init = NULL;
static tfn_shm_destroy          fn_shm_destroy = NULL;
static tfn_shm_rm_dead_queue    fn_shm_rm_dead_queue = NULL;

// shm queue create drop
static tfn_queue_create         fn_queue_create = NULL;
static tfn_queue_attach         fn_queue_attach = NULL;
static tfn_queue_drop           fn_queue_drop = NULL;

// shm queue size
static tfn_queue_size           fn_queue_size = NULL;
static tfn_queue_full           fn_queue_full = NULL;
static tfn_queue_empty          fn_queue_empty = NULL;

// shm queue push
static tfn_queue_push           fn_queue_push = NULL;
static tfn_queue_push_nowait    fn_queue_push_nowait = NULL;
static tfn_queue_push_timeout   fn_queue_push_timeout = NULL;

// shm queue pop
static tfn_queue_pop            fn_queue_pop = NULL;
static tfn_queue_pop_nowait     fn_queue_pop_nowait = NULL;
static tfn_queue_pop_timeout    fn_queue_pop_timeout = NULL;
static tfn_queue_free_poped     fn_queue_free_poped = NULL;

// socket mode
static tfn_socket_open          fn_socket_open = NULL;
static tfn_socket_close         fn_socket_close = NULL;
static tfn_socket_bind          fn_socket_bind = NULL;
static tfn_socket_listen        fn_socket_listen = NULL;
static tfn_socket_connect       fn_socket_connect = NULL;
static tfn_socket_send          fn_socket_send = NULL;
static tfn_socket_recv          fn_socket_recv = NULL;
static tfn_socket_buf_free      fn_socket_buf_free = NULL;
static tfn_socket_port          fn_socket_port = NULL;

//////////////////////////////////////////////////////////////////////

// labels
// shm
const static char l_shm_init[] = "shm_init";
const static char l_shm_destroy[] = "shm_destroy";
const static char l_shm_rm_dead_queue[] = "shm_remove_queues_exclude";

// queue
const static char l_queue_create[] = "shmqueue_create";
const static char l_queue_attach[] = "shmqueue_attach";
const static char l_queue_drop[] = "shmqueue_drop";

// queue size
const static char l_queue_size[] = "shmqueue_size";
const static char l_queue_full[] = "shmqueue_full";
const static char l_queue_empty[] = "shmqueue_empty";

// queue push
const static char l_queue_push[] = "shmqueue_push";
const static char l_queue_push_nowait[] = "shmqueue_push_nowait";
const static char l_queue_push_timeout[] = "shmqueue_push_timeout";

// queue pop
const static char l_queue_pop[] = "shmqueue_pop";
const static char l_queue_pop_nowait[] = "shmqueue_pop_nowait";
const static char l_queue_pop_timeout[] = "shmqueue_pop_timeout";
const static char l_queue_free_poped[] = "shmqueue_free";

// mode socket
const static char l_socket_open[] = "mod_open_socket";
const static char l_socket_close[] = "mod_close_socket";
const static char l_socket_bind[] = "mod_socket_bind";
const static char l_socket_listen[] = "mod_listen";
const static char l_socket_connect[] = "mod_connect";
const static char l_socket_send[] = "mod_send";
const static char l_socket_recv[] = "mod_recv";
const static char l_socket_free[] = "mod_free";
const static char l_socket_port[] = "mod_get_socket_port";


//////////////////////////////////////////////////////////////////////

// dlopen dynamic library
typedef void* hcsoftbus;
hcsoftbus c_softbus_handle(const char *so);
void c_softbus_release(hcsoftbus lib);

// shm manipulate
void wrap_fn_shm_init(hcsoftbus lib, int size);
void wrap_fn_shm_destroy(hcsoftbus lib);
void wrap_fn_shm_rm_dead_queue(hcsoftbus lib, void *array, int len);

// queue create/drop
void* wrap_fn_queue_create(hcsoftbus lib, int *key, int queue_size);
void* wrap_fn_queue_attach(hcsoftbus lib, int key);
void wrap_fn_queue_drop(hcsoftbus lib, void* shmq);

// queue size
int wrap_fn_queue_size(hcsoftbus lib, void *shmq);
int wrap_fn_queue_full(hcsoftbus lib, void *shmq);
int wrap_fn_queue_empty(hcsoftbus lib, void *shmq);

// queue push
int wrap_fn_queue_push(hcsoftbus lib, void *shmq, void *ele, int ele_size);
int wrap_fn_queue_push_nowait(hcsoftbus lib, void *shmq, void *ele, int ele_size);
int wrap_fn_queue_push_timeout(hcsoftbus lib, void *shmq, void *ele, int ele_size, int sec, int nsec);

// queue pop
int wrap_fn_queue_pop(hcsoftbus lib, void *shmq, void **ele, int *ele_size);
int wrap_fn_queue_pop_nowait(hcsoftbus lib, void *shmq, void **ele, int *ele_size);
int wrap_fn_queue_pop_timeout(hcsoftbus lib, void *shmq, void **ele, int *ele_size, int sec, int nsec);
void wrap_fn_queue_free_poped(hcsoftbus lib, void *ptr);

/////////////////////////////////////////////////////////
// socket mode
void *wrap_fn_socket_open(hcsoftbus lib, int mod);
int wrap_fn_socket_close(hcsoftbus lib, void* s);
int wrap_fn_socket_bind(hcsoftbus lib, void *s, int port);
int wrap_fn_socket_listen(hcsoftbus lib, void *s);
int wrap_fn_socket_connect(hcsoftbus lib, void *s, int port);
int wrap_fn_socket_send(hcsoftbus lib, void *s, const void *buf, const int size);
int wrap_fn_socket_recv(hcsoftbus lib, void *s, void **buf, int *size);
void wrap_fn_socket_buf_free(hcsoftbus lib, void *buf);
int wrap_fn_socket_port(hcsoftbus lib, void *s);



#ifdef __cplusplus
}
#endif

#endif

 libcsoftbus_func.h

New file
@@ -0,0 +1,153 @@
#ifndef _c_libshm_queue_so_func_h_
#define _c_libshm_queue_so_func_h_


#ifdef __cplusplus
extern "C"{
#endif

/**
 * 初始化共享内存
 * @size 共享内存大小
 *
 */
typedef void(*tfn_shm_init)(int);
/**
 * 销毁共享内存
 * 整个进程退出时需要执行这个方法，该方法首先会检查是否还有其他进程在使用该共享内存，如果还有其他进程在使用就只是detach,如果没有其他进程在使用则销毁整块内存。
 */
typedef void(*tfn_shm_destroy)();
//移除不包含在keys中的队列
typedef void (*tfn_shm_rm_dead_queue)(void *keys, int length);

/**
 * 创建队列
 * @ shmqueue
 * @ key 标识共享队列的唯一标识, key是一个指针里面存储了key的值， 如果key的值为-1系统会自动分配一个key值并把该key的值赋给key指针。如果key的值不会空会检查是否有重复绑定的情况, 有重复就报错没有就创建队列并绑定key.
 * @ queue_size 队列大小
 */
typedef void*(*tfn_queue_create) (int*, int);
/**
 * 绑定key到队列，但是并不会创建队列。如果没有对应指定key的队列提示错误并返回空值
 */
typedef void*(*tfn_queue_attach) (int);
/**
 * 销毁
*/
typedef void(*tfn_queue_drop) (void*);
/**
 * 队列元素的个数
 */
typedef int(*tfn_queue_size) (void*);
/**
 * 是否已满
 * * @return 1是， 0否
 */
typedef int(*tfn_queue_full) (void*);
/**
 * 是否为空
 * @return 1是， 0否
 */
typedef int(*tfn_queue_empty) (void*);
/**
 * 入队, 队列满时等待.
 * @return 1 入队成功, 0 入队失败
 */
typedef int(*tfn_queue_push) (void*, void*, int);
/**
 * 入队, 队列满时立即返回.
 * @return 1 入队成功, 0 入队失败
 */
typedef tfn_queue_push tfn_queue_push_nowait;
/**
 * 入队, 指定时间内入队不成功就返回
 * @sec 秒
 * @nsec 纳秒
 * @return 1 入队成功, 0 入队失败
 */
typedef int(*tfn_queue_push_timeout) (void*, void*, int, int, int);
/**
 * 出队, 队列空时等待
 * @return 1 出队成功， 0出队失败
 */
typedef int(*tfn_queue_pop) (void*, void**, int *);
/**
 * 出队, 队列空时立即返回
 * @return 1 出队成功， 0出队失败
 */
typedef tfn_queue_pop tfn_queue_pop_nowait;
/**
 * 出队, 指定时间内出队不成功就返回
 * @sec秒
 * @nsec纳秒
 * @return 1 出队成功， 0出队失败
 */
typedef int(*tfn_queue_pop_timeout) (void*, void**, int *, int, int);
/**
 * 释放出队分配的内存
 */
typedef void (*tfn_queue_free_poped)(void *ptr);

////////////////////////////
// socket mode
////////////////////////////
enum socket_mod_t
{
    PULL_PUSH = 1,
    REQ_REP = 2,
    PAIR = 3,
    PUB_SUB = 4,
    SURVEY = 5,
    BUS = 6

};

/**
 * 创建socket
 * @return socket地址
*/
typedef void*(*tfn_socket_open) (int);
/**
 * 关闭socket
*/
typedef int(*tfn_socket_close) (void*);
/**
 * 绑定端口到socket, 如果不绑定则系统自动分配一个
 * @return 0 成功， 其他值 失败的错误码
*/
typedef int(*tfn_socket_bind) (void*, int);
/**
 * 服务端开启连接监听
 * @return 0 成功， 其他值 失败的错误码
 */
typedef int(*tfn_socket_listen) (void*);
/**
 * 客户端发起连接请求
 */
typedef int(*tfn_socket_connect) (void*, int);
/**
 * 发送信息
 * @return 0 成功， 其他值 失败的错误码
 */
typedef int(*tfn_socket_send) (void*, const void*, const int);
/**
 * 接收信息
 * @return 0 成功， 其他值 失败的错误码
*/
typedef int(*tfn_socket_recv) (void*, void**, int*);

/**
 * 释放接收信息的buf
 */
typedef void(*tfn_socket_buf_free) (void*);

/**
 * 获取soket端口号
 */
typedef int(*tfn_socket_port) (void*);

#ifdef __cplusplus
}
#endif

#endif

 softbus.go

New file
@@ -0,0 +1,259 @@
package softbus

/*
#cgo LDFLAGS: -ldl
#include <stdlib.h>
#include "libcsoftbus.h"

void *get_array(const int size){
    int *arr = (int*)malloc(sizeof(int)*size);
    return arr;
}

void set_array(void *arr, const int index, const int value){
    int *iarr = (int*)arr;
    iarr[index] = value;
}
*/
import "C"
import (
    "errors"
    "unsafe"
)

var libsoftbus C.hcsoftbus

// Init Dynamic library
func Init(so string) error {
    cso := C.CString(so)
    defer C.free(unsafe.Pointer(cso))

    handle := C.c_softbus_handle(cso)
    if handle == nil {
        // fmt.Errorf("Go SoftBus Init From %s Error\n", so)
        return errors.New("Init SoftBus Error")
    }

    libsoftbus = handle
    return nil
}

// Release handle of c softbus
func Release() {
    if libsoftbus != nil {
        C.c_softbus_release(libsoftbus)
    }
}

// ShmInit block size
func ShmInit(size int) error {
    if libsoftbus == nil {
        return errors.New("C SoftBus Handle Is Nil")
    }
    C.wrap_fn_shm_init(libsoftbus, C.int(size))
    return nil
}

// ShmDestroy destroy shm block, every softbus proc MUST call it
func ShmDestroy() {
    if libsoftbus != nil {
        C.wrap_fn_shm_destroy(libsoftbus)
    }
}

// ShmRmDeadQueue remove dead queue but pass live queue keys
func ShmRmDeadQueue(keys []int) error {
    if libsoftbus == nil {
        return errors.New("C SoftBus Handle Is Nil")
    }
    if len(keys) == 0 {
        return errors.New("Live Queue Keys Is Nil")
    }

    l := len(keys)
    ca := C.get_array(C.int(l))
    defer C.free(ca)
    for k, v := range keys {
        C.set_array(ca, C.int(k), C.int(v))
    }

    C.wrap_fn_shm_rm_dead_queue(libsoftbus, ca, C.int(l))
    return nil
}

// Queue struct wrap shmqueue pointer
type Queue struct {
    shmqueue unsafe.Pointer
    key      int
}

// Create shm queue
func Create(qSize int) *Queue {
    if libsoftbus == nil {
        return nil
    }

    var key C.int
    csb := C.wrap_fn_queue_create(libsoftbus, &key, C.int(qSize))
    if csb != nil {
        return &Queue{
            shmqueue: csb,
            key:      int(key),
        }
    }
    return nil
}

// Attach shm queue with key
func Attach(key int) *Queue {
    if libsoftbus == nil {
        return nil
    }

    csb := C.wrap_fn_queue_attach(libsoftbus, C.int(key))
    if csb != nil {
        return &Queue{
            shmqueue: csb,
            key:      key,
        }
    }

    return nil
}

// Drop shm queue
func (q *Queue) Drop() error {
    if libsoftbus == nil {
        return errors.New("C SoftBus Handle Is Nil")
    }

    C.wrap_fn_queue_drop(libsoftbus, q.shmqueue)
    return nil
}

// Size shm queue
func (q *Queue) Size() int {
    if libsoftbus == nil {
        return 0
    }

    cs := C.wrap_fn_queue_size(libsoftbus, q.shmqueue)
    return int(cs)
}

// Full shm queue
func (q *Queue) Full() bool {
    if libsoftbus == nil {
        return true
    }

    f := C.wrap_fn_queue_full(libsoftbus, q.shmqueue)
    if f != 0 {
        return true
    }
    return false
}

// Empty shm queue
func (q *Queue) Empty() bool {
    if libsoftbus == nil {
        return false
    }

    f := C.wrap_fn_queue_full(libsoftbus, q.shmqueue)
    if f != 0 {
        return true
    }
    return false

}

// Push shm queue
func (q *Queue) Push(data []byte) error {
    if libsoftbus == nil {
        return errors.New("C SoftBus Handle Is Nil")
    }

    r := C.wrap_fn_queue_push(libsoftbus, q.shmqueue, unsafe.Pointer(&data[0]), C.int(len(data)))
    if r == 0 {
        return nil
    }
    return errors.New("Queue Push Data Error")
}

// PushNoWait shm queue
func (q *Queue) PushNoWait(data []byte) error {
    if libsoftbus == nil {
        return errors.New("C SoftBus Handle Is Nil")
    }

    r := C.wrap_fn_queue_push_nowait(libsoftbus, q.shmqueue, unsafe.Pointer(&data[0]), C.int(len(data)))
    if r == 0 {
        return nil
    }
    return errors.New("Queue PushNoWait Data Error")
}

// PushTimeout shm queue push with timeout
func (q *Queue) PushTimeout(data []byte, sec, nsec int) error {
    if libsoftbus == nil {
        return errors.New("C SoftBus Handle Is Nil")
    }

    r := C.wrap_fn_queue_push_timeout(libsoftbus, q.shmqueue, unsafe.Pointer(&data[0]), C.int(len(data)), C.int(sec), C.int(nsec))
    if r == 0 {
        return nil
    }
    return errors.New("Queue PushTimeout Data Error")
}

// Pop from shm queue
func (q *Queue) Pop() ([]byte, error) {
    if libsoftbus == nil {
        return nil, errors.New("C SoftBus Handle Is Nil")
    }

    var data unsafe.Pointer
    var size C.int
    r := C.wrap_fn_queue_pop(libsoftbus, q.shmqueue, &data, &size)
    if r == 0 {
        gd := C.GoBytes(data, size)
        C.wrap_fn_queue_free_poped(libsoftbus, data)
        return gd, nil
    }
    return nil, errors.New("Queue Pop Data Error")
}

// PopNoWait from shm queue
func (q *Queue) PopNoWait() ([]byte, error) {
    if libsoftbus == nil {
        return nil, errors.New("C SoftBus Handle Is Nil")
    }

    var data unsafe.Pointer
    var size C.int
    r := C.wrap_fn_queue_pop_nowait(libsoftbus, q.shmqueue, &data, &size)
    if r == 0 {
        gd := C.GoBytes(data, size)
        C.wrap_fn_queue_free_poped(libsoftbus, data)
        return gd, nil
    }
    return nil, errors.New("Queue Pop Data Error")
}

// PopTimeout from shm queue
func (q *Queue) PopTimeout(sec, nsec int) ([]byte, error) {
    if libsoftbus == nil {
        return nil, errors.New("C SoftBus Handle Is Nil")
    }

    var data unsafe.Pointer
    var size C.int
    r := C.wrap_fn_queue_pop_timeout(libsoftbus, q.shmqueue, &data, &size, C.int(sec), C.int(nsec))
    if r == 0 {
        gd := C.GoBytes(data, size)
        C.wrap_fn_queue_free_poped(libsoftbus, data)
        return gd, nil
    }
    return nil, errors.New("Queue Pop Data Error")
}

 softbusSocket.go

New file
@@ -0,0 +1,127 @@
package softbus

/*
#include <stdlib.h>
#include "libcsoftbus.h"
*/
import "C"
import (
    "errors"
    "fmt"
    "unsafe"
)

const (

    // RETVAL retual val
    RETVAL = 0x7fff
    // PullPush mode
    PullPush = 1
    // ReqRep mode
    ReqRep = 2
    // Pair mode
    Pair = 3
    // PubSub mode
    PubSub = 4
    // Survey mode
    Survey = 5
    // Bus mode
    Bus = 6
)

// Socket struct
type Socket struct {
    csocket unsafe.Pointer
}

// OpenSocket open
func OpenSocket(mod int) *Socket {
    if libsoftbus == nil {
        return nil
    }

    s := C.wrap_fn_socket_open(libsoftbus, C.int(mod))
    if s == nil {
        return nil
    }

    return &Socket{
        csocket: s,
    }
}

// Close socket
func (s *Socket) Close() int {
    if libsoftbus != nil || s.csocket != nil {
        r := C.wrap_fn_socket_close(libsoftbus, s.csocket)
        return int(r)
    }
    return RETVAL
}

// Bind socket
func (s *Socket) Bind(port int) int {
    if libsoftbus == nil {
        return RETVAL
    }

    r := C.wrap_fn_socket_bind(libsoftbus, s.csocket, C.int(port))
    return int(r)
}

// Listen socket
func (s *Socket) Listen() int {
    if libsoftbus == nil {
        return RETVAL
    }
    r := C.wrap_fn_socket_listen(libsoftbus, s.csocket)
    return int(r)
}

// Connect socket
func (s *Socket) Connect(port int) int {
    if libsoftbus == nil {
        return RETVAL
    }

    r := C.wrap_fn_socket_connect(libsoftbus, s.csocket, C.int(port))
    return int(r)
}

// Send data
func (s *Socket) Send(data []byte) int {
    if libsoftbus == nil {
        return RETVAL
    }

    r := C.wrap_fn_socket_send(libsoftbus, s.csocket, unsafe.Pointer(&data[0]), C.int(len(data)))
    return int(r)
}

// Recv data
func (s *Socket) Recv() ([]byte, error) {
    if libsoftbus == nil {
        return nil, errors.New("libsoftbus is nil")
    }

    var rd unsafe.Pointer
    var rl C.int
    r := C.wrap_fn_socket_recv(libsoftbus, s.csocket, &rd, &rl)
    if r != 0 {
        return nil, fmt.Errorf("Recv Data Failed errCode: %d", int(r))
    }
    data := C.GoBytes(rd, rl)
    C.wrap_fn_socket_buf_free(libsoftbus, rd)
    return data, nil
}

// Port get port/key
func (s *Socket) Port() int {
    if libsoftbus == nil {
        return -1
    }

    r := C.wrap_fn_socket_port(libsoftbus, s.csocket)

    return int(r)
}