-
IPV4流媒体C语言项目
IPV4流媒体项目
项目分析：基于UDP的IPV4流媒体广播，主要广播mp3文件，根据李慧芹老师的课学习完成。
代码下载：放最后了（包括QT实现客户端）服务器、客户端分析
服务器主要完成内容：
1）参数初始化，命令行设置
2）网络配置
3）守护进程设置
4）节目单获取
5）节目单发送线程
6）各频道节目发送线程
客户端主要完成内容：
1）参数初始化，命令行设置
2）网络配置
3）接收节目单 选择频道
4）父进程读取频道数据 写给子进程
5）子进程读取父进程数据 交给媒体解析器解析共用库文件
客户端和服务器数据交换文件，便是共用库文件来源：proto.h，主要包含频道包结构体和节目单包结构体
#ifndef PROTO_H__ #define PROTO_H__ #define DEFAULT_MGROUP "224.2.2.2" #define DEFAULT_RCVPORT "6666" #define CHNNR 100 #define LISTCHNID 0 #define MINCHNID 1 #define MAXCHNID (CHNNR+MINCHNID -1) #define MSG_CHANNEL_MAX (65536 -20 -8) //推荐长度 -ip包的长度 - udp包的报头 #define MAX_DATA (MSG_CHANNEL_MAX - sizeof(chnid_t)) #define MSG_LIST_MAX (65536 -20 -8) //推荐长度 -ip包的长度 - udp包的报头 #define MAX_ENTRY (MSG_LIST_MAX - sizeof(chnid_t)) #include "sitetype.h" /* 一个频道包的结构体 */ struct msg_channel_st { chnid_t chinid; uint8_t data[1]; }__attribute__((packed)); /* 节目包结构体 */ struct msg_listentry_st { chnid_t chnid; /* channel num */ uint16_t len; uint8_t desc[1]; /* describe channel information*/ }__attribute__((packed)); /* 总包结构体 包括节目单和节目描述*/ struct msg_list_st { chnid_t chnid; /* channel num */ struct msg_listentry_st entry[1]; }__attribute__((packed)); #endif- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
其中包含的sitetype.h是为了隐藏实际用的数据类型
#ifndef SITETYPE_H__ #define SITETYPE_H__ #includetypedef uint8_t chnid_t; #endif - 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
服务器
主函数文件：（根据服务器要完成的各个步骤分步骤完成），其中节目单获取部分我与李慧芹老师讲的写的不同，我直接使用的是发送节目单的结构体
struct msg_list_st，李慧芹老师使用的是struct mlib_listentry_st，各函数功能和代码注释已在代码中标明。
server.c 如下：#include#include #include "proto.h" #include "server.h" #include #include #include #include #include #include #include #include #include #include #include #include #include /* superset of previous */ #include #include "../include/mlib.h" #include "list.h" #include "channel.h" #include /* 需要释放的malloc空间个数 */ #define PTRMAXNUM 10 /* * -M Specify a multicast group * -P designated port * -F Front desk operation * -D designate media library location * -H display help information * * */ /* 初始化参数配置*/ struct server_conf_st server_conf = { .rcvport = DEFAULT_RCVPORT, .mgroup = DEFAULT_MGROUP, .runmode = RUN_DAEMON, .media_dir = DEFAULT_MEDIADIR, .ifname = DEFAULT_IF }; int sfd; struct ip_mreqn mreqn; struct sockaddr_in sndaddr; void *freeptr[PTRMAXNUM]; int free_ptrnum; static void printf_help() { printf(" * -M --mgroup 指定多播组\n\ * -P --port 指定接受端口\n\ * -F --player 指定前台运行\n\ * -D 指定媒体库文件目录\n\ * -H --help 显示帮助* \n"); } /* 把当前进程设置为守护进程，一般为固定步骤 */ static int daemonize() // create daemon process { pid_t pid; int fd; pid = fork(); if (pid < 0) { syslog(LOG_ERR, "fork():%s", strerror(errno)); perror("fork()"); exit(1); } if (pid > 0) { exit(0); } fd = open("/dev/null", O_RDWR); if (fd < 0) { syslog(LOG_ERR, "open():%s", strerror(errno)); perror("open()"); exit(1); } dup2(fd, 0); dup2(fd, 1); dup2(fd, 2); //脱离标准终端 if (fd > 2) //防止内存泄漏 { close(fd); } setsid(); if (chdir("/") < 0) //把守护进程指定到根目录 { syslog(LOG_ERR, "chdir()%s", strerror(errno)); } umask(0); return 0; } /* 异常退出，之后执行线程收尸 结构体指针释放 */ static void daemon_exit() //用于被信号终止之后调用 { for(int i = 0; i < free_ptrnum; i++) { free(freeptr[i]); } thr_list_destroy(); syslog(LOG_INFO, "daemon_exit():%s", strerror(errno)); closelog(); close(sfd); exit(0); } /* 网络设置函数 */ static int socket_init() { sfd = socket(AF_INET, SOCK_DGRAM, 0); if (sfd < 0) { syslog(LOG_ERR, "socket():%s", strerror(errno)); exit(1); } inet_pton(AF_INET, server_conf.mgroup, &mreqn.imr_multiaddr); inet_pton(AF_INET, "0.0.0.0", &mreqn.imr_address); mreqn.imr_ifindex = if_nametoindex("ens33"); if (setsockopt(sfd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreqn, sizeof(mreqn)) < 0) //man 7 ip 把当前sdf加入到多播组 { syslog(LOG_ERR, "setsockopt():%s", strerror(errno)); exit(1); } sndaddr.sin_family = AF_INET; sndaddr.sin_port = htons(atoi(server_conf.rcvport)); //字序一定要转换 inet_pton(AF_INET, server_conf.mgroup, &sndaddr.sin_addr.s_addr); return 0; } int main(int argc, char **argv) { int c; /* 异常信号处理 */ struct sigaction sa; //设置要处理的信号集合 sa.sa_handler = daemon_exit; //处理函数 sigemptyset(&sa.sa_mask); //清空信号集里的信号 sigaddset(&sa.sa_mask, SIGQUIT); //把指定信号加入信号集 sigaddset(&sa.sa_mask, SIGTERM); sigaddset(&sa.sa_mask, SIGINT); sigaction(SIGTERM, &sa, NULL); // 对指定信号处理 sigaction(SIGINT, &sa, NULL); sigaction(SIGQUIT, &sa, NULL); openlog("netradio", LOG_PID | LOG_PERROR, LOG_DAEMON); //打开系统日志 /*1) cmd _ analyse*/ while (1) { c = getopt(argc, argv, "M:P:FD:H"); if (c < 0) { break; } switch (c) { case 'M': server_conf.mgroup = optarg; break; case 'P': server_conf.rcvport = optarg; break; case 'F': server_conf.runmode = RUN_FOREGROUND; break; case 'D': server_conf.media_dir = optarg; break; case 'H': printf_help(); exit(0); break; default: abort(); break; } } /* 2) initialize socket */ socket_init(); /* 3) come ture daemon process */ if (server_conf.runmode == RUN_DAEMON) { if (daemonize() != 0) { perror("daemonize"); } } else if (server_conf.runmode == RUN_FOREGROUND) { } else { syslog(LOG_ERR, "EINVAL server_conf.runmode."); fprintf(stderr, "EINVAL\n"); exit(1); } /* 4) obtain channel information*/ struct msg_list_st *list = NULL; //节目单总列表 int list_num = 0; //节目单列表里频道个数 int err; err = mlib_getchnlist(&list, &list_num); //获取回填 if (err) { syslog(LOG_ERR, "mlib_getchnlist()%s", strerror(errno)); exit(1); } struct msg_listentry_st *ptr; #ifdef DEBUG printf("list_num = %d\n", list_num); /*****变长结构体的一般处理方式,不能使用list[i]进行访问********************/ int i = 0; for (ptr = list->entry; i < list_num; ptr = (void *)((char *)ptr) + ntohs(ptr->len)) { i++; printf("channel:%d:%s", ptr->chnid, ptr->desc); } #endif // DEBUG /* 5) create program list pthread*/ err = thr_list_create(list, list_num); if (err) { syslog(LOG_ERR, "thr_list_create()%s", strerror(errno)); exit(1); } /* create chanel pthread*/ int i; i = 0; for (ptr = list->entry; i < list_num; ptr = (void *)((char *)ptr) + ntohs(ptr->len)) { i++; err = thr_channel_create(ptr); if (err) { syslog(LOG_ERR, "thr_channel_create()%s", strerror(errno)); exit(1); } } freeptr[0] = ptr; free_ptrnum = 1; while (1) { pause(); } } - 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
- 128
- 129
- 130
- 131
- 132
- 133
- 134
- 135
- 136
- 137
- 138
- 139
- 140
- 141
- 142
- 143
- 144
- 145
- 146
- 147
- 148
- 149
- 150
- 151
- 152
- 153
- 154
- 155
- 156
- 157
- 158
- 159
- 160
- 161
- 162
- 163
- 164
- 165
- 166
- 167
- 168
- 169
- 170
- 171
- 172
- 173
- 174
- 175
- 176
- 177
- 178
- 179
- 180
- 181
- 182
- 183
- 184
- 185
- 186
- 187
- 188
- 189
- 190
- 191
- 192
- 193
- 194
- 195
- 196
- 197
- 198
- 199
- 200
- 201
- 202
- 203
- 204
- 205
- 206
- 207
- 208
- 209
- 210
- 211
- 212
- 213
- 214
- 215
- 216
- 217
- 218
- 219
- 220
- 221
- 222
- 223
- 224
- 225
- 226
- 227
- 228
- 229
- 230
- 231
- 232
- 233
- 234
- 235
- 236
- 237
- 238
- 239
- 240
- 241
- 242
- 243
- 244
- 245
- 246
- 247
- 248
- 249
- 250
- 251
- 252
- 253
- 254
- 255
- 256
- 257
- 258
- 259
- 260
server.c，主要是大框架设置，比较简单
项目主要难点在于节目单列表和频道的获取，在mlib.c中完成获取节目单
在获取节目单函数
int mlib_getchnlist(struct msg_list_st **mchnarr, int *index)中，先解析频道目录的个数，申请当前频道数需要的最大内存entrylistptr = malloc(MSG_LIST_MAX/CHNNR*globres.gl_pathc);，然后在根据path2entry函数解析到的频道为节目单结构体进行赋值，需要注意的是由于结构体是变长结构体，不可以直接用下标数组访问、赋值等，具体做法见代码。
mlib.c：#include#include #include "sitetype.h" #include "mlib.h" #include #include #include "mytbf.h" #include #include "proto.h" #include "server.h" #include #include #include #include #include #include"mytbf.h" #include"server.h" #define PATHSIZE 1024 #define LINEBUFSIZE 1024 #define MP3_BITRATE 320 * 1024 // 采样率 struct channel_context_st { chnid_t chnid; char *desc; glob_t mp3glob; int pos; int fd; off_t offset; mytbf_t *tbf; }; struct channel_context_st channel[CHNNR + 1]; int totalsize; struct channel_context_st *path2entry(char *gl_pathv) { printf("current path: %s\n", gl_pathv); syslog(LOG_INFO, "current path: %s\n", gl_pathv); char pathstr[PATHSIZE] = {'\0'}; char linebuf[LINEBUFSIZE]; FILE *fp; struct channel_context_st *me; static chnid_t cur_id = MINCHNID; // 由于是一个静态变量所以相当于一直在操作同一块内存 有叠加效果 strcat(pathstr, gl_pathv); //拼接两个字符串 strcat(pathstr, "/desc.txt"); fp = fopen(pathstr, "r"); // 打开频道描述文件 syslog(LOG_INFO, "channel dir:%s\n", pathstr); if (fp == NULL) { syslog(LOG_INFO, "%s is not a channe; dir (can not find desc.txt)", gl_pathv); return NULL; } if (fgets(linebuf, LINEBUFSIZE, fp) == NULL) { syslog(LOG_INFO, "%s is not a channel dir(cant get the desc.text)", gl_pathv); fclose(fp); return NULL; } fclose(fp); // 关闭频道描述文件 me = malloc(sizeof(*me)); if (me == NULL) { syslog(LOG_ERR, "malloc()%s", strerror(errno)); return NULL; } me->tbf = mytbf_init(MP3_BITRATE / 8, MP3_BITRATE / 8 * 5); if (me->tbf == NULL) { syslog(LOG_ERR, "mytbf_init()%s\n", strerror(errno)); return NULL; } me->desc = strdup(linebuf); strncpy(pathstr, gl_pathv, PATHSIZE); strncat(pathstr, "/*.mp3", PATHSIZE - 1); if (glob(pathstr, 0, NULL, &me->mp3glob) != 0) { // cur_id++; syslog(LOG_ERR, "%s is not a channel dir(can not find mp3 files", gl_pathv); free(me); return NULL; } me->pos = 0; me->offset = 0; me->fd = open(me->mp3glob.gl_pathv[me->pos], O_RDONLY); if (me->fd < 0) { syslog(LOG_WARNING, "%s open failed.", me->mp3glob.gl_pathv[me->pos]); free(me); return NULL; } me->chnid = cur_id; cur_id++; return me; } int mlib_freechnlist(struct msg_listentry_st *ptr) { free(ptr); return 0; } /* 根据目录解析频道 */ int mlib_getchnlist(struct msg_list_st **mchnarr, int *index) { char path[PATHSIZE]; glob_t globres; int size = 0; int i; int num = 0; /****************换msg_listentry_st************************************/ struct msg_listentry_st *ptr = NULL; struct msg_list_st *entrylistptr; //节目单总包 struct channel_context_st *res; for (i = 0; i < CHNNR + 1; i++) { channel[i].chnid = -1; } snprintf(path, PATHSIZE, "%s/*", server_conf.media_dir); if (glob(path, 0, NULL, &globres) < 0) { return -1; } totalsize = sizeof(chnid_t); /*****************************malloc**************************/ entrylistptr = malloc(MSG_LIST_MAX/CHNNR*globres.gl_pathc); entrylistptr->chnid = LISTCHNID; if(entrylistptr == NULL) { syslog(LOG_ERR, "malloc() failed.%s", strerror(errno)); exit(1); } ptr = entrylistptr->entry; for (i = 0; i < globres.gl_pathc; i++) { // globres.gl_pathv[i] -> "/var/media/ch1" res = path2entry(globres.gl_pathv[i]); if (res != NULL) { syslog(LOG_DEBUG, "path2entry () :%d %s\n", res->chnid, res->desc); totalsize += sizeof(struct msg_listentry_st)+strlen(res->desc); // ptr = realloc(ptr, totalsize); /* （1）sizeof一般是用来求某种类型的变量所占用的内存大小的。 （2）strlen顾名思义，是用来求string的length的。 */ size = sizeof(struct msg_listentry_st)+strlen(res->desc); //如果用sizeof(res->decs)算出来的是指针的大小 会丢失数据 ptr->chnid = res->chnid; ptr->len = htons(size); strcpy((char *) ptr->desc, res->desc); ptr = (void *)(((char *) ptr) + size); // 向后移动ptr /* 把可以用的频道结构体保存 */ memcpy(channel + res->chnid, res, sizeof(*res)); //加res->chnid 代表第几个频道 与&channel[res->chnid]效果相同 num++; #ifdef DEBUG printf("sizeof(*res)%d\n", sizeof(*res)); #endif // DEBUG } } *mchnarr = realloc(entrylistptr, totalsize); //重新分配内存大小 if (*mchnarr == NULL) { syslog(LOG_ERR, "realloc() failed."); } *index = num; #ifdef DEBUG printf("totalsize = %d\n", totalsize); #endif // DEBUG return 0; } static int open_next(chnid_t chnid) { for (int i = 0; i < channel[chnid].mp3glob.gl_pathc; ++i) { channel[chnid].pos++; // 更新下一个媒体文件偏移 if (channel[chnid].pos == channel[chnid].mp3glob.gl_pathc) { syslog(LOG_INFO, "没有文件了\n"); channel[chnid].pos = 0; } close(channel[chnid].fd); // 尝试打开新文件 channel[chnid].fd = open(channel[chnid].mp3glob.gl_pathv[channel[chnid].pos], O_RDONLY); if (channel[chnid].fd < 0) { syslog(LOG_WARNING, "open(%s):%s", channel[chnid].mp3glob.gl_pathv[chnid], strerror(errno)); } else { syslog(LOG_INFO, "打开新的文件\n"); channel[chnid].offset = 0; return 0; } } syslog(LOG_ERR, "None of mp3 in channel %d id available.", chnid); return -1; } ssize_t mlib_readchn(chnid_t chnid, void *buf, size_t size) { int tbfsize; int len; // get token number tbfsize = mytbf_fetchtoken(channel[chnid].tbf, size); syslog(LOG_INFO, "current tbf():%d\n", mytbf_checktoken(channel[chnid].tbf)); while (1) { /* 从fd中读取tbfsize大小的数据到buf中，读取地址=文件开始+offset*/ len = pread(channel[chnid].fd, buf, tbfsize, channel[chnid].offset); // 读取tbfsize数据到从offset处开始的buf /*current song open failed*/ if (len < 0) { // 当前这首歌可能有问题，错误不至于退出，读取下一首歌 syslog(LOG_WARNING, "media file %s pread():%s", channel[chnid].mp3glob.gl_pathv[channel[chnid].pos], strerror(errno)); if(open_next(chnid) <0) { syslog(LOG_ERR, "channel %d no successed open",channel[chnid].chnid); } } else if (len == 0) { syslog(LOG_DEBUG, "media %s file is over", channel[chnid].mp3glob.gl_pathv[channel[chnid].pos]); #ifdef DEBUG printf("current chnid :%d\n", chnid); #endif open_next(chnid); break; } else /*len > 0*/ //真正读取到了数据 { channel[chnid].offset += len; break; } } // remain some token if (tbfsize - len > 0) mytbf_returntoken(channel[chnid].tbf, tbfsize - len); printf("current chnid :%d\n", chnid); return len; //返回读取到的长度 } - 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
- 128
- 129
- 130
- 131
- 132
- 133
- 134
- 135
- 136
- 137
- 138
- 139
- 140
- 141
- 142
- 143
- 144
- 145
- 146
- 147
- 148
- 149
- 150
- 151
- 152
- 153
- 154
- 155
- 156
- 157
- 158
- 159
- 160
- 161
- 162
- 163
- 164
- 165
- 166
- 167
- 168
- 169
- 170
- 171
- 172
- 173
- 174
- 175
- 176
- 177
- 178
- 179
- 180
- 181
- 182
- 183
- 184
- 185
- 186
- 187
- 188
- 189
- 190
- 191
- 192
- 193
- 194
- 195
- 196
- 197
- 198
- 199
- 200
- 201
- 202
- 203
- 204
- 205
- 206
- 207
- 208
- 209
- 210
- 211
- 212
- 213
- 214
- 215
- 216
- 217
- 218
- 219
- 220
- 221
- 222
- 223
- 224
- 225
- 226
- 227
- 228
- 229
- 230
- 231
- 232
- 233
- 234
- 235
- 236
- 237
- 238
- 239
- 240
- 241
- 242
- 243
- 244
- 245
- 246
- 247
- 248
- 249
- 250
- 251
- 252
- 253
- 254
- 255
- 256
- 257
- 258
- 259
- 260
- 261
- 262
- 263
- 264
- 265
- 266
- 267
- 268
节目单发送线程
在获得节目单列表list后直接交给节目单发送线程发送即可:
list.c#include#include #include "proto.h" #include "server.h" #include #include #include #include #include #include #include #include #include #include #include #include #include /* superset of previous */ #include #include "mlib.h" #include "list.h" #include "channel.h" #include struct msg_list_st *list_entry; int num_list_entry; pthread_t list_tid; struct msg_list_st *entrylistptr; //节目单总包 struct msg_listentry_st *entryptr; //每一个节目的包 void *thr_list(void *thr) { int ret; struct msg_listentry_st *ptr; #ifdef DEBUG int i = 0; for (ptr = list_entry->entry; i < num_list_entry; ptr = (void *)((char *)ptr) + ntohs(ptr->len)) { i++; printf("channel:%d:%s", ptr->chnid, ptr->desc); } #endif // DEBUG while (1) { syslog(LOG_INFO, "thr_list sndaddr :%d\n", sndaddr.sin_addr.s_addr); ret = sendto(sfd, list_entry, totalsize, 0, (void *)&sndaddr, sizeof(sndaddr)); // 频道列表在广播网段每秒发送entrylist syslog(LOG_DEBUG, "List sent content len:%d\n", totalsize); if (ret < 0) { syslog(LOG_WARNING, "sendto(serversd, enlistp...:%s", strerror(errno)); } else { syslog(LOG_DEBUG, "List sendto(serversd, enlistp....):success"); } sleep(1); } } int thr_list_create(struct msg_list_st *listptr, int num_ent) { int err; list_entry = listptr; //接收节目单的包 所有的节目单包的首地址 num_list_entry = num_ent; //频道的个数 err = pthread_create(&list_tid, NULL, thr_list, NULL); if (err) { syslog(LOG_ERR, "pthread_create():%s\n", strerror(errno)); return -1; } return 0; } // 销毁节目单线程 int thr_list_destroy(void) { pthread_cancel(list_tid); pthread_join(list_tid, NULL); free(entrylistptr); return 0; } - 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
频道发送线程
频道发送和节目单发送类似，也是在mlib.c 解析需要的的数据，分给线程发送，
解析频道数据在mlib_readchn(chnid_t chnid, void *buf, size_t size)函数中完成，主要完成把mp3文件打开回填数据到发送buf中（即频道包结构体里的data中），然后发送给客户端，发送线程代码如下：
channel.c#include#include #include #include #include #include #include "proto.h" #include "list.h" #include "mlib.h" #include "server.h" #include "server.h" #include #include #include #include /* superset of previous */ int num_list_entry; static int tid_nextpos = 0; struct thr_channel_entry_st { chnid_t chnid; pthread_t tid; }; struct thr_channel_entry_st thr_channel[CHNNR]; void *thr_channel_snder(void *ptr) { struct msg_channel_st *sbufp; int len; struct msg_listentry_st *entry = ptr; sbufp = malloc(MSG_CHANNEL_MAX); if (sbufp == NULL) { syslog(LOG_ERR, "malloc():%s", strerror(errno)); exit(1); } sbufp->chinid = entry->chnid; // 频道号处理 printf("sbufp->chinid %d\n", sbufp->chinid); // 频道内容读取 while (1) { len = mlib_readchn(entry->chnid, sbufp->data, 320 * 1024 / 8); // 320kbit/s 读取某一个频道的媒体文件 syslog(LOG_DEBUG, "mlib_readchn() len: %d", len); if (len < 0) { break; } if (sendto(sfd, sbufp, len + sizeof(chnid_t), 0, (void *)&sndaddr, sizeof(sndaddr)) < 0) { syslog(LOG_ERR, "thr_channel(%d):sendto():%s", entry->chnid, strerror(errno)); break; } syslog(LOG_ERR, "thr_channel()%d:sendto():%s", entry->chnid, strerror(errno)); sched_yield(); //主动出让调度器 } pthread_exit(NULL); } int thr_channel_create(struct msg_listentry_st *list) { int err; err = pthread_create(&thr_channel[tid_nextpos].tid, NULL, thr_channel_snder, list); if (err) { syslog(LOG_ERR, "pthread_create():%s\n", strerror(errno)); return -1; } thr_channel[tid_nextpos].chnid = list->chnid; //填写频道信息 tid_nextpos++; return 0; } int thr_channel_destroy(struct mlib_listentry_st *ptr) { int i; for (i = 0; i < CHNNR; i++) { if (thr_channel[i].chnid == ptr->chnid) { if (pthread_cancel(thr_channel[i].tid) < 0) { syslog(LOG_ERR, "pthread_cancel():thr thread of channel%d", ptr->chnid); return -ESRCH; } pthread_join(thr_channel[i].tid, NULL); thr_channel[i].chnid = -1; break; } } return 0; } int thr_channel_destroyall(void) { for (int i = 0; i < CHNNR; i++) { if (thr_channel[i].chnid > 0) { if (pthread_cancel(thr_channel[i].tid) < 0) { syslog(LOG_ERR, "pthread_cancel():thr thread of channel:%ld", thr_channel[i].tid); return -ESRCH; } pthread_join(thr_channel[i].tid, NULL); thr_channel[i].chnid = -1; } } return 0; } - 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
令牌桶发送流量控制,与之前讲的没什么区别
mytbf.c#include#include #include #include #include #include #include #include #include #include #include #include"pthread.h" #include "mytbf.h" static int min(int a, int b) { return a < b ? a : b; } struct mytbf_st { int cps; // c per second int burst; int token; int pos; pthread_mutex_t mut; pthread_cond_t cond; }; static struct mytbf_st *job[MYTBF_MAX]; static pthread_mutex_t mut_job = PTHREAD_MUTEX_INITIALIZER; static pthread_once_t once_init = PTHREAD_ONCE_INIT; static pthread_t tid; static void alrm_handle(int sig) { pthread_mutex_lock(&mut_job); for (int i = 0; i < MYTBF_MAX; ++i) { if (job[i] != NULL) { pthread_mutex_lock(&job[i]->mut); job[i]->token += job[i]->cps; if (job[i]->token > job[i]->burst) { job[i]->token = job[i]->burst; } pthread_cond_broadcast(&job[i]->cond); // 惊群 pthread_mutex_unlock(&job[i]->mut); } } pthread_mutex_unlock(&mut_job); } static void *thr_alrm(void *p) { struct itimerval tick; memset(&tick, 0, sizeof(tick)); tick.it_value.tv_sec = 1; // sec tick.it_value.tv_usec = 0; // micro sec. tick.it_interval.tv_sec = 1; tick.it_interval.tv_usec = 0; signal(SIGALRM, alrm_handle); setitimer(ITIMER_REAL, &tick, NULL); while (1) { pause(); } return 0; } // 模块卸载函数 static void module_unload() { int i; pthread_cancel(tid); pthread_join(tid, NULL); for (i = 0; i < MYTBF_MAX; i++) { free(job[i]); } return; } // 模块加载函数 static void module_load() { int err; err = pthread_create(&tid, NULL, thr_alrm, NULL); if (err) { fprintf(stderr, "pthread_create():%s", strerror(errno)); exit(1); } atexit(module_unload); } static int get_free_pos_unlocked() { for (int i = 0; i < MYTBF_MAX; ++i) { if (job[i] == NULL) { return i; } } return -1; } // 初始化一个令牌桶 mytbf_t *mytbf_init(int cps, int burst) { struct mytbf_st *me; module_load(); // 开启定时token派发 pthread_once(&once_init, module_load); // 限定只开启一次 int pos; // 初始化mytbf me = malloc(sizeof(*me)); if (me == NULL) { return NULL; } me->cps = cps; me->burst = burst; me->token = 0; pthread_mutex_init(&me->mut, NULL); // 初始化该令牌桶的mutex pthread_cond_init(&me->cond, NULL); // 初始化该令牌桶的conditional variable pthread_mutex_lock(&mut_job); pos = get_free_pos_unlocked(); if (pos < 0) { pthread_mutex_unlock(&mut_job); fprintf(stderr, "no free position,\n"); free(me); exit(1); } me->pos = pos; job[me->pos] = me; // 分配槽位 pthread_mutex_unlock(&mut_job); return me; } int mytbf_fetchtoken(mytbf_t *ptr, int size) { int n; struct mytbf_st *me = ptr; pthread_mutex_lock(&me->mut); while (me->token <= 0) pthread_cond_wait(&me->cond, &me->mut); // 没有令牌的时候 等待信号量通知 n = min(me->token, size); me->token -= n; pthread_cond_broadcast(&me->cond); pthread_mutex_unlock(&me->mut); return n; } int mytbf_returntoken(mytbf_t *ptr, int size) { struct mytbf_st *me = ptr; pthread_mutex_lock(&me->mut); me->token += size; if (me->token > me->burst) me->token = me->burst; pthread_mutex_unlock(&me->mut); return 0; } int mytbf_destroy(mytbf_t *ptr) { struct mytbf_st *me = ptr; pthread_mutex_lock(&mut_job); job[me->pos] = NULL; pthread_mutex_unlock(&mut_job); pthread_mutex_destroy(&me->mut); pthread_cond_destroy(&me->cond); free(ptr); return 0; } int mytbf_checktoken(mytbf_t *ptr) { int token_left = 0; struct mytbf_st *me = ptr; pthread_mutex_lock(&me->mut); token_left = me->token; pthread_mutex_unlock(&me->mut); return token_left; } - 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
- 128
- 129
- 130
- 131
- 132
- 133
- 134
- 135
- 136
- 137
- 138
- 139
- 140
- 141
- 142
- 143
- 144
- 145
- 146
- 147
- 148
- 149
- 150
- 151
- 152
- 153
- 154
- 155
- 156
- 157
- 158
- 159
- 160
- 161
- 162
- 163
- 164
- 165
- 166
- 167
- 168
- 169
- 170
- 171
- 172
- 173
- 174
- 175
- 176
客户端
相对简单，主要完成以下内容：
1）参数初始化，命令行设置
2）网络配置加入多播组
3）接收节目单 选择频道
4）父进程读取频道数据 写给子进程
5）子进程读取父进程数据 交给媒体解析器解析
client.c 和 client.h如下：#include#include #include #include "client.h" #include "proto.h" #include #include #include #include #include #include #include #include #include /* * -M --mgroup 指定多播组 * -P --port 指定接受端口 * -p --player 指定播放器 * -H --help 显示帮助 * */ #define DEFAULT_PLAYER "/usr/bin/mpg123 - > /dev/null" #define IPSIZE 30 struct client_conf_st client_conf = { .rcvport = DEFAULT_RCVPORT, .mgroup = DEFAULT_MGROUP, .player_cmd = DEFAULT_PLAYER}; static int sfd; void printf_help() { printf(" * -M --mgroup 指定多播组\n\ * -P --port 指定接受端口\n\ * -p --player 指定播放器\n\ * -H --help 显示帮助* \n"); } int writen(int fd, const void *buf, size_t nbyte) { int len; int write_len = 0; if (nbyte > 0) { while (1) { len = write(fd, buf, nbyte); if (len < 0) { if (len = -EINTR) { continue; } else { perror("write()"); exit(1); } } if (len == 0) { break; } nbyte = nbyte - len; write_len += len; } } return write_len; } int main(int argc, char **argv) { /* *初始化 * 级别：默认值 配置文件 环境变量 命令行参数 * * */ int pd[2]; int c; int index; pid_t pid; uint64_t receive_buf_size = 1024; struct ip_mreqn mreqn; struct option argarr[] = { {"port", 1, NULL, 'p'}, {"mgroup", 1, NULL, 'M'}, {"player", 1, NULL, 'p'}, {"help", 1, NULL, 'H'}, {NULL, 0, NULL, 0}}; struct sockaddr_in laddr, seraddr, raddr; socklen_t seraddr_len, raddrlen; // while (1) // { // c = getopt_long(argc, argv, "M:P:p:H", argarr, &index); //从命令行里搜索这些关键词 并且后面有参数 // printf("c = %d\n", c); // c = 0; // if (c < 0) // { // break; // } // switch (c) // { // case 'P': // client_conf.rcvport = optarg; // break; // case 'M': // client_conf.mgroup = optarg; // break; // case 'p': // client_conf.player_cmd = optarg; // break; // case 'H': // printf_help(); // break; // default: // abort(); // break; // } // } /* 网络配置 */ sfd = socket(AF_INET, SOCK_DGRAM, 0); if (sfd < 0) { perror("socket()"); exit(1); } inet_pton(AF_INET, client_conf.mgroup, &mreqn.imr_multiaddr); //IPv4点分式转二进制数 把地址填入mreqn.imr_multiaddr inet_pton(AF_INET, " 0.0.0.0", &mreqn.imr_address); mreqn.imr_ifindex = if_nametoindex("ens33"); if (setsockopt(sfd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreqn, sizeof(mreqn)) < 0) //man 7 ip 把当前sdf加入到多播组 { perror("setsockopt()"); exit(1); } int val; val = 1; if (setsockopt(sfd, IPPROTO_IP, IP_MULTICAST_LOOP, &(val), sizeof(val)) < 0) { perror("setsockopt()"); exit(1); } // improve efficiency uint64_t receive_buff_size = 20 * 1024 * 1024; //20M if (setsockopt(sfd, SOL_SOCKET, SO_RCVBUF, &receive_buff_size, sizeof(receive_buff_size)) < 0) { perror("setsockopt()"); exit(1); } laddr.sin_family = AF_INET; laddr.sin_port = htons(atoi(client_conf.rcvport)); //主机序转网络序 inet_pton(AF_INET, "0.0.0.0", &laddr.sin_addr); if (bind(sfd, (void *)&laddr, sizeof(laddr)) < 0) { perror("bind()"); exit(1); } /* 进程通信管道 */ if (pipe(pd) < 0) { perror("pipe()"); exit(1); } pid = fork(); if (pid < 0) { perror("fork()"); exit(1); } /* 子进程读管道解析 */ if (pid == 0) { close(sfd); close(pd[1]); dup2(pd[0], 0); //把管道的读端复制作为0号文件描述符 if (pd[0] > 0) //如果的管道读端不是标准输入，关闭 { close(pd[0]); } execl("/bin/sh", "sh", "-c", client_conf.player_cmd, NULL); perror("execl"); exit(0); } else //父进程 读网络数据 写管道 { char ipstr_raddr[IPSIZE], ipstr_server_addr[IPSIZE]; //收节目单 struct msg_list_st *msg_list; msg_list = malloc(MSG_LIST_MAX); if (msg_list == NULL) { perror("malloc"); exit(1); } int len; seraddr_len = sizeof(seraddr); while (1) { len = recvfrom(sfd, msg_list, MSG_LIST_MAX, 0, (void *)&seraddr, &seraddr_len); if (len < 0) { perror("recvform()"); exit(1); } if (len < (int)sizeof(msg_list)) { fprintf(stderr, "message is too small\n"); continue; } if (msg_list->chnid != LISTCHNID) { fprintf(stderr, "chinnal id is not match\n"); continue; } break; } //打印节目单 选择频道 struct msg_listentry_st *pos; for (pos = msg_list->entry; (char *)pos < ((char *)msg_list + len); pos = (void *)((char *)pos) + ntohs(pos->len)) { printf("channel:%d:%s", pos->chnid, pos->desc); } int ret = 0; int chosenid; while (ret < 1) { ret = scanf("%d", &chosenid); if (ret != 1) exit(1); } //收频道包，发给子进程 struct msg_channel_st *msg_channel; msg_channel = malloc(MSG_CHANNEL_MAX); if (msg_channel == NULL) { perror("malloc()"); exit(1); } raddrlen = sizeof(raddr); while (1) { len = recvfrom(sfd, msg_channel, MSG_CHANNEL_MAX, 0, (void *)&raddr, &raddrlen); if (raddr.sin_addr.s_addr != seraddr.sin_addr.s_addr) //防止其他ip恶意发包 { /*******把IP的大整数转化为字符串************/ inet_ntop(AF_INET, &raddr.sin_addr.s_addr, ipstr_raddr, 30); inet_ntop(AF_INET, &seraddr.sin_addr.s_addr, ipstr_server_addr, 30); fprintf(stderr, "Ignore:addr not match. raddr:%s server_addr:%s.\n", ipstr_raddr, ipstr_server_addr); continue; } if (len < sizeof(msg_channel)) { fprintf(stderr, "msg_channel too small\n"); continue; } if (msg_channel->chnid == chosenid) { fprintf(stdout, "accepted msg : %s \n", msg_channel->data); writen(pd[1], msg_channel->data, len - sizeof(chnid_t)); // sleep(2); } } close(sfd); free(msg_channel); } } - 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
- 128
- 129
- 130
- 131
- 132
- 133
- 134
- 135
- 136
- 137
- 138
- 139
- 140
- 141
- 142
- 143
- 144
- 145
- 146
- 147
- 148
- 149
- 150
- 151
- 152
- 153
- 154
- 155
- 156
- 157
- 158
- 159
- 160
- 161
- 162
- 163
- 164
- 165
- 166
- 167
- 168
- 169
- 170
- 171
- 172
- 173
- 174
- 175
- 176
- 177
- 178
- 179
- 180
- 181
- 182
- 183
- 184
- 185
- 186
- 187
- 188
- 189
- 190
- 191
- 192
- 193
- 194
- 195
- 196
- 197
- 198
- 199
- 200
- 201
- 202
- 203
- 204
- 205
- 206
- 207
- 208
- 209
- 210
- 211
- 212
- 213
- 214
- 215
- 216
- 217
- 218
- 219
- 220
- 221
- 222
- 223
- 224
- 225
- 226
- 227
- 228
- 229
- 230
- 231
- 232
- 233
- 234
- 235
- 236
- 237
- 238
- 239
- 240
- 241
- 242
- 243
- 244
- 245
- 246
- 247
- 248
- 249
- 250
- 251
- 252
- 253
- 254
- 255
- 256
- 257
- 258
- 259
- 260
- 261
- 262
- 263
- 264
- 265
- 266
- 267
- 268
- 269
- 270
- 271
- 272
- 273
- 274
- 275
client.h
#ifndef CLIENT_H__ #define CLIENT_H__ #include#include #include #define DEFAULT_PLAYERCMD "/usr/bin/mpg123 - > /dev/null" //mpg123 - 表示从标准输入读区文件进行解析 struct client_conf_st { char *rcvport; char *mgroup; char *player_cmd; }; extern struct client_conf_st client_conf; #endif - 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
代码：https://download.csdn.net/download/qq_43745917/87167158
服务器QT客户端：git地址
-
相关阅读:
Hadoop+Hive数据分析综合案例
[附源码]Python计算机毕业设计Django个性化名片网站
拓端tecdat|R语言时间序列分解和异常检测方法应用案例
conda 创建虚拟环境
Tomcat多实例部署实验
基于javaweb的农业信息管理系统(java+ssm+jsp+js+html+mysql)
IP RAN基站回传中的三大组网方案
UTF-8和UTF-16扩展方案
IOS停机卡免流线路下载更新
Android服务器——TomCat服务器的搭建
- 原文地址：https://blog.csdn.net/qq_43745917/article/details/128048044