HEXA开发日志目录
上一篇 HEXA娱乐开发日志技术点005——死而复生之Gstreamer推流

---

前言

更新

虽然没人看，但是还是要给自己一个说法。好久没有更新了，但是我并没有停止这个项目，只是从大步跑变成了小步慢跑，不信你看下图（L和G都是我，名字不同是因为不同主机的签名不同，后面我可能会改一下），我还是每周改了一点东西的，图中这个项目在码云上，只是用来装暂时代码的，就不公开了。
这次虽然更新了，但是没有实际进展，就是把最近的失败尝试总结一下。

困难

进展缓慢并不是遇到了什么大难题，只是一些很繁琐的麻烦。

1. 交叉编译好烦啊
   上回说到我用Gstreamer推流成功，下面要整合到skill内部，但是skill的编译在上位机进行，用Gstreamer需要用到Gstreamer的库和头文件，这个在上位机编译环境中没有。
   正面刚这个问题有两个办法，一是学习docker，在docker容器（上位机）里面交叉编译出Gstreamer库之后，保存这个容器为镜像，用这个镜像替代编译skill的镜像；二是直接把板子上的Gstreamer库和头文件放到docker容器里，更新镜像。
   当然还有其它方法，例如研究skill的编译过程，自己登陆到docker容器里把它搞定等。
   总之是没有姿势漂亮又学习成本低的方法。
2. 进程通信好烦啊
   正面刚不行就迂回解决。我把程序分为了2部分，一部分控制Gstreamer，独立成一个程序，暂且叫它G server吧，另一部分在skill中，暂且叫它G client吧。server只管控制Gstreamer，直接在HEXA体内编译，client只管发送命令，不用到Gstreamer的东西，就可以直接在现在的上位机编译了。
   思路很简单，就是实现有点繁琐，包括信令制定和调试，当然最大的问题还在于我对这类程序的不熟悉产生的抗拒。唉~想想就头大。

日拱一卒

linux进程通信

要两个程序相互配合，自然要进程通信，在linux环境下进程通信的方法百度都有，我不展开。我选的是使用套接字的方法，示例代码网上也一大堆。

协议和信令制定

我的协议和信令都是随便制定的，并没有参考优秀代码，因为平时接触不多，也不知道要参考什么，跟着感觉做吧。

协议

协议就是大家对话的语言规则，机器不会说人话，只认字节，协议给这些字节一个解释方法和发言规则，大家才能沟通。
我的协议制定如下面代码，只有协议版本和信令两部分，发言规则就是client一句命令，server会有一个ACK回应。

struct protocol {
    unsigned short version;
    //must be last member
    struct signal signal;
};

信令

信令这个词是我从一起拿接触到的东西里拿来用的，它也是给字节流一个解释方法，相对于协议，信令像是一个字典，只可以用来对照理解意思，而协议还要规范大家的说话顺序。

信令类型(enum eSignalType)	参数
播放	播放地址url
停止	无
ACK	错误码enum result
写字（未实现）	文字+位置+大小

信令就是下面这个结构体

struct signal {
    unsigned long length;//data长度
    unsigned long type;//信令类型
    unsigned long cmdId;//信令/命令编号，client每次发送+1，用来区分不同的发送
    unsigned char data[0];//柔性数组装参数
};

拱出的代码

因为是我暂存代码库的代码，都很糙，仅供参考。

client端

主要就4个函数，open_client和close_client分别是和套接字接力和解除绑定，start_client和stop_client分别发送播放和停止信令，最后main对它们进行测试。

#include <sys/types.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#define UNIX_DOMAIN "/tmp/DDM.domain"
#define PROTOCOL_VERSION 0
#define RECEIVE_MAX 1024
enum result {
    RESULT_OK,
    RESULT_ERR_FAIL,
    RESULT_ERR_VERSION,
    RESULT_ERR_LENGTH,
};

enum eSignalType{
    E_SIGNAL_START,
    E_SIGNAL_STOP,
    E_SIGNAL_ACK,
};

struct signal {
    unsigned long length;
    unsigned long type;
    unsigned long cmdId;
    unsigned char data[0];
};

struct protocol {
    unsigned short version;
    //must be last member
    struct signal signal;
};
static int connect_fd = -1;
static struct sockaddr_un srv_addr = {
    .sun_family = AF_UNIX,
    .sun_path = UNIX_DOMAIN
};

static void dump(struct protocol *p){
    int i;
    printf("++++\n");
    printf("version=%u\n",p->version);
    printf("signal.length=%lu\n",p->signal.length);
    printf("signal.type=%lu\n",p->signal.type);
    printf("signal.cmdId=%lu\n",p->signal.cmdId);
    for (i = 0; i < p->signal.length; ++i){
        printf("%u ", p->signal.data[i]);
    }
    printf("\n");
    printf("----\n");
}

int open_client(void)
{
    int ret = 0;
    printf("opening client.\n");
    if (connect_fd < 0)
    {
        connect_fd = socket(PF_UNIX,SOCK_STREAM,0);
        if(connect_fd < 0){
            printf("%s\n", strerror(errno));
            printf("creat socket error.\n");
            return connect_fd;
        }

        ret = connect(connect_fd, (struct sockaddr*)&srv_addr, sizeof(srv_addr));
        if (ret < 0){
            printf("%s\n", strerror(errno));
            printf("connect server error.\n");
            close(connect_fd);
            connect_fd = -1;
            return ret;
        }
        printf("open client success.\n");
    }
    return ret;
}

void close_client(void)
{
    printf("closing client.\n");
    close(connect_fd);
    connect_fd = -1;
    printf("close client success.\n");
}
int i;
static unsigned int cmdId = 0;
int start_client(char *url)
{
    int ret = -1;
    struct protocol *pdata, data = {
        .version = PROTOCOL_VERSION,
        .signal = {
            .type = E_SIGNAL_START,
        },
    };
    int length = strlen(url);
    if (length <= 0 || length > 512){
        printf("url error");
        printf("%s\n", url);
        return -1;
    }
    data.signal.cmdId = cmdId++;
    data.signal.length = length + 1;
    pdata = (struct protocol *)malloc(sizeof(struct protocol) + data.signal.length);
    *pdata = data;
    memcpy(pdata->signal.data, url, length + 1);
    printf("starting client.\n");
    //dump(pdata);
    if (write(connect_fd, pdata, sizeof(struct protocol) + data.signal.length) == sizeof(struct protocol) + data.signal.length){
        if (read(connect_fd, pdata, sizeof(struct protocol) + data.signal.length) > 0){
            //printf("[%s %d]%d\n", __FUNCTION__, __LINE__);
            //dump(pdata);
            if (pdata->signal.cmdId == data.signal.cmdId && pdata->signal.type == E_SIGNAL_ACK){
                ret = 0;
                printf("got ACK error=%d\n", pdata->signal.data[0]);
            }
        }
        else{
            printf("[%s %d]read error\n", __FUNCTION__, __LINE__);
        }
    }
    free(pdata);
    if (ret){
        printf("fail to start client:");
    }
    else{
        printf("started client:");
    }
    printf("%s\n", url);
    return ret;
}

int stop_client(void)
{
    int ret = -1;
    struct protocol *pdata, data = {
        .version = PROTOCOL_VERSION,
        .signal = {
            .type = E_SIGNAL_STOP,
        },
    };
    data.signal.cmdId = cmdId++;
    data.signal.length = 0;
    pdata = (struct protocol *)malloc(sizeof(struct protocol) + data.signal.length);
    *pdata = data;
    printf("stopping client.\n");
    //dump(pdata);
    if (write(connect_fd, pdata, sizeof(struct protocol) + data.signal.length) == sizeof(struct protocol) + data.signal.length){
        if (read(connect_fd, pdata, sizeof(struct protocol) + data.signal.length) > 0){
            //printf("[%s %d]%d\n", __FUNCTION__, __LINE__);
            //dump(pdata);
            if (pdata->signal.cmdId == data.signal.cmdId && pdata->signal.type == E_SIGNAL_ACK){
                ret = 0;
                printf("got ACK error=%d\n", pdata->signal.data[0]);
            }
        }
        else{
            printf("[%s %d]read error\n", __FUNCTION__, __LINE__);
        }
    }
    free(pdata);
    if (ret){
        printf("fail to stop client\n");
    }
    else{
        printf("stopped client\n");
    }
    return ret;
}
#include <unistd.h>

int main(){
    int ret;
    pid_t pid;
    pid = fork();
    if (pid == 0){
        ret = system("./simple_server");
        printf("[%s %d] %d\n", __FUNCTION__, __LINE__, ret);
    }
    else{
        sleep(1);
        ret = open_client();
        printf("[%s %d] %d\n", __FUNCTION__, __LINE__, ret);
        if (!ret){
            ret = start_client("please start!");
            printf("[%s %d] %d\n", __FUNCTION__, __LINE__, ret);
            ret = stop_client();
            printf("[%s %d] %d\n", __FUNCTION__, __LINE__, ret);
            close_client();
        }
    }
}

server端

NOT_IN_HEXA中包的代码是和Gstreamer相关的，把这部分注释起来是为了在任意类UNIX平台上调试。

#include<stdio.h>
#include<sys/socket.h>
#include<sys/types.h>
#include<sys/un.h>
#include<stdlib.h>
#include<unistd.h>
#include<pthread.h>
#include<string.h>

#define NOT_IN_HEXA
#ifndef NOT_IN_HEXA
#include <gst/gst.h>
#include <glib.h>
#endif

#define UNIX_DOMAIN "/tmp/DDM.domain"
#define PROTOCOL_VERSION 0
#define RECEIVE_MAX 1024

enum result {
    RESULT_OK,
    RESULT_ERR_FAIL,
    RESULT_ERR_VERSION,
    RESULT_ERR_LENGTH,
};

enum eSignalType{
    E_SIGNAL_START,
    E_SIGNAL_STOP,
    E_SIGNAL_ACK,
};

struct signal {
    unsigned long length;
    unsigned long type;
    unsigned long cmdId;
    unsigned char data[0];
};

struct protocol {
    unsigned short version;
    //must be last member
    struct signal signal;
};

static struct sockaddr_un srv_addr = {
    .sun_family = AF_UNIX,
    .sun_path = UNIX_DOMAIN
};

static pthread_mutex_t mutex;
static int running = 0;
#ifndef NOT_IN_HEXA
static GMainLoop *loop = NULL;
static GstElement *pipeline, *videosrc, *text, *videoenc, *videoconvert, *muxer, *sink;
static GstBus *bus;
static void *start_loop(){
    /* Iterate */
    g_print ("Running...\n");
    running = 1;
    g_main_loop_run (loop);
    running = 0;
    return NULL;
}
#endif
static void dump(struct protocol *p){
    int i;
    printf(">>>>\n");
    printf("version=%u\n",p->version);
    printf("signal.length=%lu\n",p->signal.length);
    printf("signal.type=%lu\n",p->signal.type);
    printf("signal.cmdId=%lu\n",p->signal.cmdId);
    for (i = 0; i < p->signal.length; ++i){
        printf("%u ", p->signal.data[i]);
    }
    printf("\n");
    printf("<<<<\n");
}
static enum result Start(int argc, char *argv[]){
    int ret = 0;
#ifndef NOT_IN_HEXA
    pthread_t id;
    /* Initialisation */
    pthread_mutex_lock(&mutex);
    {
        if (loop != NULL){
            pthread_mutex_unlock(&mutex);
        return RESULT_ERR_FAIL;
        }
        gst_init (&argc, &argv);
        loop = g_main_loop_new (NULL, FALSE);
    }
    pthread_mutex_unlock(&mutex);
    if (argc < 3)
        return RESULT_ERR_FAIL;
    /* Create gstreamer elements */
    pipeline = gst_pipeline_new ("media-player");
    videosrc = gst_element_factory_make ("v4l2src",         "video-camrta-source");
    text     = gst_element_factory_make ("textoverlay",         "text");
    videoenc = gst_element_factory_make ("imxvpuenc_h264",  "video-h264-byte-stream");
    videoconvert = gst_element_factory_make ("h264parse",       "video-convert");
    muxer    = gst_element_factory_make ("flvmux",          "flv-muxer");
    sink     = gst_element_factory_make ("rtmpsink",      "sink");

    if (!pipeline || !videosrc || !text || !videoenc || !videoconvert || !muxer || !sink) {
        g_printerr ("One element could not be created. Exiting.\n");
        loop = NULL;
        return RESULT_OK;
    }

    /* Set up the pipeline */
    /* we set the input filename to the source element */
    g_object_set (G_OBJECT (sink), "location", argv[1], NULL);
    g_object_set (G_OBJECT (text), "text", argv[2], NULL);
    /* we add a message handler */

    /* we add all elements into the pipeline */
    gst_bin_add_many (GST_BIN (pipeline), videosrc, text, videoenc, videoconvert, muxer, sink, NULL);
    /* we link the elements together */
    if (gst_element_link (videosrc, text)){
        g_print ("link success %d\n", __LINE__);
    }
    else{
        return -1;
    }
    if (gst_element_link (text, videoenc)){
        g_print ("link success %d\n", __LINE__);
    }
    else{
        return -1;
    }
    if (gst_element_link (videoenc, videoconvert)){
        g_print ("link success %d\n", __LINE__);
    }
    else{
        return -1;
    }
    if (gst_element_link (videoconvert, muxer)){
        g_print ("link success %d\n", __LINE__);
    }
    else{
        return -1;
    }
    if (gst_element_link (muxer, sink)){
        g_print ("link success %d\n", __LINE__);
    }
    else{
        return -1;
    }

    /* Set the pipeline to "playing" state*/
    //g_print ("Now playing: %s\n", argv[1]);
    gst_element_set_state (pipeline, GST_STATE_PLAYING);
    ret = pthread_create(&id, NULL, (void*)start_loop, NULL);
#endif
    if (ret){
        printf("create thread fail %d\n", __LINE__);
        return RESULT_ERR_FAIL;
    }
    return RESULT_OK;
}

static enum result Stop(){
#ifndef NOT_IN_HEXA
    pthread_mutex_lock(&mutex);
    {
        int count = 0;
        if (loop == NULL){
            pthread_mutex_unlock(&mutex);
            return RESULT_ERR_FAIL;
        }
        g_main_loop_quit(loop);
        while (running && count < 10000) count++;
        g_print ("Returned, stopping playback\n");
        gst_element_set_state (pipeline, GST_STATE_NULL);

        g_print ("Deleting pipeline\n");
        gst_object_unref (GST_OBJECT (pipeline));
        g_main_loop_unref (loop);
        loop = NULL;
    }
    pthread_mutex_unlock(&mutex);
#endif
    return RESULT_OK;
}
int i;

void process_rcv(int client_fd, char *rcv_buff, ssize_t len) {
    ssize_t num;
    //parse
    struct protocol *data = (struct protocol*)rcv_buff;
    struct protocol *pfeedback, feedbackData = {
        .version = PROTOCOL_VERSION,
        .signal = {
            .type = E_SIGNAL_ACK,
        },
    };
    enum result error = RESULT_OK;
    if (!len) return;
    if (!data) {
        error = RESULT_ERR_FAIL;
    }
    if (data->version != PROTOCOL_VERSION) {
        error = RESULT_ERR_VERSION;
    }
    if (error == RESULT_OK
      && ((data->signal.length + sizeof(struct protocol)) > RECEIVE_MAX
        || data->signal.length + sizeof(struct protocol) != len
        || len > RECEIVE_MAX)) {
        printf("[%s %d] %zd\n", __FUNCTION__, __LINE__, len);
        error = RESULT_ERR_LENGTH;
    }
    //printf("[%s %d] %d\n", __FUNCTION__, __LINE__, error);
    //dump(data);
    //deal with
    if (error == RESULT_OK) {
        switch ((enum eSignalType)data->signal.type)
        {
            case E_SIGNAL_START:
                error = Start(1, NULL);
            break;
            case E_SIGNAL_STOP:
                error = Stop();
            break;
            default:
                error = RESULT_ERR_FAIL;
            break;
        }
    }
    //feedback
    feedbackData.signal.cmdId = data->signal.cmdId;
    feedbackData.signal.length = sizeof(unsigned long);
    pfeedback = (struct protocol *)malloc(sizeof(struct protocol) + feedbackData.signal.length);
    if (pfeedback)
    {
        memcpy(pfeedback, &feedbackData, sizeof(feedbackData));
        ((unsigned long *)pfeedback->signal.data)[0] = (unsigned long)error;
        //printf("[%s %d]]\n", __FUNCTION__, __LINE__);
        //dump(pfeedback);
        num = write(client_fd, pfeedback, sizeof(struct protocol) + feedbackData.signal.length);
        printf("sent ACK for cmd %lu with result %u\n", data->signal.cmdId, error);
        free(pfeedback);
    }
}
int main() {
    int server_fd, client_fd;
    int ret = 0;
    ssize_t num = 1;
    int i;
    struct sockaddr_un client_addr;
    socklen_t addrlen;
    server_fd = socket(AF_UNIX, SOCK_STREAM, 0);
    if(server_fd < 0){
        perror("connect creat communication socket");
        return server_fd;
    }
    unlink(UNIX_DOMAIN);

    ret = bind(server_fd, (struct sockaddr*)&srv_addr, sizeof(srv_addr));
    if (ret < 0) {
        perror("cannot bind server socket");
        goto exit;
    }

    ret = listen(server_fd, 1);
    if (ret < 0) {
        perror("cannot listen sockfd");
        goto exit;
    }

    pthread_mutex_init(&mutex, NULL);
    printf("server started\n");
    client_fd = accept(server_fd, NULL, &addrlen);
    if (client_fd < 0){
        perror("cannot accept requst");
        goto exit;
    }

    while (num){
        char rcv_buff[1024];
        memset(rcv_buff, 0, sizeof(rcv_buff));
        num = read(client_fd, rcv_buff, sizeof(rcv_buff));
        process_rcv(client_fd, rcv_buff, num);
    }
    close(client_fd);
exit:
    close(server_fd);
    unlink(UNIX_DOMAIN);
    return ret;
}

拱歪了

其实这点东西都很经典了，没什么难度的，拱歪的地方在于一句permission denied，只要用skill跑这些代码，server的bind函数和client的connect都会报这个错，server我可以用root手动跑起来，但是client我就没办法了。
因为我的调试都是在HEXA的root下运行的，不会有权限问题，而且在MAC电脑上调试也没有需要这个权限。HEXA关于权限的控制在这里，能加的都加了还是不行。
好在linux的进程通信还有其它方法，这个思路应该是可以的。

---

下一篇 未出炉