[asterisk-commits] phsultan: branch phsultan/rtmp-support r272520 - in /team/phsultan/rtmp-suppo...
SVN commits to the Asterisk project
asterisk-commits at lists.digium.com
Fri Jun 25 07:37:48 CDT 2010
Author: phsultan
Date: Fri Jun 25 07:37:41 2010
New Revision: 272520
URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=272520
Log:
Simplify this code by introducing APIs.
New APIs : librtmp, libavformat
- librtmp handles RTMP connections to any server, so now, we should be able
to connect to FMS, Red5, Wowza, etc.. It also provides traffic securing
protocols (RTMPS, RTMPE), that we might implement in the future.
librtmp is part of rtmpdump : http://rtmpdump.mplayerhq.hu/
- libavformat is not currently used. It will help us stream audio/video files
in any format, and play them back to SIP/H323 peers. libavformat is part of
FFMPEG.
New configuration options have been introduced as well to set/unset debugging
options available in librtmp.
Modified:
team/phsultan/rtmp-support/build_tools/menuselect-deps.in
team/phsultan/rtmp-support/channels/chan_rtmp.c
team/phsultan/rtmp-support/configure
team/phsultan/rtmp-support/configure.ac
team/phsultan/rtmp-support/include/asterisk/autoconfig.h.in
team/phsultan/rtmp-support/makeopts.in
Modified: team/phsultan/rtmp-support/build_tools/menuselect-deps.in
URL: http://svnview.digium.com/svn/asterisk/team/phsultan/rtmp-support/build_tools/menuselect-deps.in?view=diff&rev=272520&r1=272519&r2=272520
==============================================================================
--- team/phsultan/rtmp-support/build_tools/menuselect-deps.in (original)
+++ team/phsultan/rtmp-support/build_tools/menuselect-deps.in Fri Jun 25 07:37:41 2010
@@ -1,5 +1,6 @@
ALSA=@PBX_ALSA@
AVCODEC=@PBX_AVCODEC@
+AVFORMAT=@PBX_AVFORMAT@
BLUETOOTH=@PBX_BLUETOOTH@
CRYPTO=@PBX_CRYPTO@
BISON=@PBX_BISON@
@@ -46,6 +47,7 @@
RESAMPLE=@PBX_RESAMPLE@
AIS=@PBX_AIS@
RADIUS=@PBX_RADIUS@
+RTMP=@PBX_RTMP@
LAUNCHD=@PBX_LAUNCHD@
SPANDSP=@PBX_SPANDSP@
SPEEX=@PBX_SPEEX@
@@ -58,6 +60,7 @@
OPENSSL=@PBX_OPENSSL@
SUPPSERV=@PBX_SUPPSERV@
SYSLOG=@PBX_SYSLOG@
+SWSCALE=@PBX_SWSCALE
TONEZONE=@PBX_TONEZONE@
UNIXODBC=@PBX_UNIXODBC@
USB=@PBX_USB@
Modified: team/phsultan/rtmp-support/channels/chan_rtmp.c
URL: http://svnview.digium.com/svn/asterisk/team/phsultan/rtmp-support/channels/chan_rtmp.c?view=diff&rev=272520&r1=272519&r2=272520
==============================================================================
--- team/phsultan/rtmp-support/channels/chan_rtmp.c (original)
+++ team/phsultan/rtmp-support/channels/chan_rtmp.c Fri Jun 25 07:37:41 2010
@@ -27,6 +27,7 @@
/*** MODULEINFO
<depend>avcodec</depend>
+ <depend>rtmp</depend>
***/
#include "asterisk.h"
@@ -39,8 +40,9 @@
#include <fcntl.h>
#include <sys/ioctl.h>
#include <libavcodec/avcodec.h>
-
-#define REF_DEBUG 1
+#include <librtmp/rtmp.h>
+#include <librtmp/log.h>
+
#include "asterisk/astobj2.h"
#include "asterisk/lock.h"
#include "asterisk/channel.h"
@@ -51,19 +53,18 @@
#include "asterisk/strings.h"
#include "asterisk/app.h"
-#include "asterisk/rtmp.h"
-
-
-struct rtmp_client {
- enum rtmp_state state;
- int fd;
- pthread_t thread;
-} *client = NULL;
+#define RTMP_DEFAULT_PORT 1935
+
static struct sockaddr_in rtmpserver;
static char rtmpserverstr[50];
-static int port = 1935;
-static char application[50];
+static int port = RTMP_DEFAULT_PORT;
+static char application[200];
+static unsigned int rtmpinputrate = 11000; /* default for Nellymoser ASAO */
+static unsigned int astinputrate = 8000;
+FILE *rtmplog = 0;
+static int rlog = 0;
+static char rtmplogfile[] = "/tmp/rtmplog.txt";
static const char tdesc[] = "RTMP driver";
static const char config[] = "rtmp.conf";
@@ -74,17 +75,16 @@
static char type[] = "RTMP";
/**
- * This structure stores information about an RTMP connection :
- * - the number of streams (FLEX NetStream objects) that form the Asterisk
- * channel. A minimum of 2 streams are used per Asterisk channel to
- * receive/send data from/to the RTMP server.
- * - the name of the stream that Asterisk will publish to the RTMP server
- * - the name of the stream that Asterisk will retrieve from the RTMP server.
- * If numstreams is higher than 2, Asterisk will ask the RTMP server to play
- * streams named 'readstream-0', 'readstream-1', etc.
+ * This structure stores information about the bi-directionnal media connection
+ * to the server.
+ *
+ * A minimum of 2 connections are used per Asterisk channel to receive/send
+ * data from/to the RTMP server. Each connection is used to either publish
+ * or read a live stream to the RTMP server.
*/
struct rtmp_pvt {
struct ast_channel *owner;
+ pthread_t thread;
AVCodec *encoder;
AVCodec *decoder;
AVCodecContext *encoding_context;
@@ -94,18 +94,13 @@
unsigned int rtmpinputrate; /* default : 11000 Hz */
unsigned int astinputrate; /* default : 8000 Hz */
- /* Each stream is a member of a group of at least 2 streams :
- * - the first stream carries data to be published to the RTMP server
- * - each subsequent stream handles data (audio/video) coming
- * from the RTMP server
- *
- * numstreams contains the number of streams contained in the group
- */
- uint32_t streamid;
- int numstreams;
- int readstream_index;
char readstream[AST_MAX_EXTENSION];
char writestream[AST_MAX_EXTENSION];
+
+ /* two RTMP connections : one read a stream from the server, the other
+ * to publish a stream */
+ RTMP *rtmpin;
+ RTMP *rtmpout;
/* \brief Pipe file descriptor handles array.
* Read from pipe[0], write to pipe[1]
@@ -113,186 +108,17 @@
int pipe[2];
};
-#ifdef LOW_MEMORY
-static int hash_streamgroups_size = 17;
-static int hash_rtmpmessages_size = 17;
-#else
-static int hash_streamgroups_size = 563;
-static int hash_rtmpmessages_size = 563;
-#endif
-
-struct ao2_container *streamgroups;
-
-#define rtmp_pvt_lock(x) ao2_lock(x)
-#define rtmp_pvt_trylock(x) ao2_trylock(x)
-#define rtmp_pvt_unlock(x) ao2_unlock(x)
-
-/*! \brief
- * when we create or delete references, make sure to use these
- * functions so we keep track of the refcounts.
- * To simplify the code, we allow a NULL to be passed to streamgroup_unref().
- */
-#ifdef REF_DEBUG
-#define streamgroup_ref(arg1,arg2) streamgroup_ref_debug((arg1),(arg2), __FILE__, __LINE__, __PRETTY_FUNCTION__)
-#define streamgroup_unref(arg1,arg2) streamgroup_unref_debug((arg1),(arg2), __FILE__, __LINE__, __PRETTY_FUNCTION__)
-#define rtmpmessage_ref(arg1,arg2) rtmpmessage_ref_debug((arg1),(arg2), __FILE__, __LINE__, __PRETTY_FUNCTION__)
-#define rtmpmessage_unref(arg1,arg2) rtmpmessage_unref_debug((arg1),(arg2), __FILE__, __LINE__, __PRETTY_FUNCTION__)
-
-static struct rtmp_pvt *streamgroup_ref_debug(struct rtmp_pvt *p, char *tag, char *file, int line, const char *func)
-{
- if (p)
- __ao2_ref_debug(p, 1, tag, file, line, func);
- else
- ast_log(LOG_ERROR, "Attempt to Ref a null pointer\n");
- return p;
-}
-
-static struct rtmp_pvt *streamgroup_unref_debug(struct rtmp_pvt *p, char *tag, char *file, int line, const char *func)
-{
- if (p)
- __ao2_ref_debug(p, -1, tag, file, line, func);
- return NULL;
-}
-
-static struct rtmp_message *rtmpmessage_ref_debug(struct rtmp_message *p, char *tag, char *file, int line, const char *func)
-{
- if (p)
- __ao2_ref_debug(p, 1, tag, file, line, func);
- else
- ast_log(LOG_ERROR, "Attempt to Ref a null pointer\n");
- return p;
-}
-
-static struct rtmp_message *rtmpmessage_unref_debug(struct rtmp_message *p, char *tag, char *file, int line, const char *func)
-{
- if (p)
- __ao2_ref_debug(p, -1, tag, file, line, func);
- return NULL;
-}
-#else
-static struct rtmp_pvt *streamgroup_ref(struct rtmp_pvt *p, char *tag)
-{
- if (p)
- ao2_ref(p, 1);
- else
- ast_log(LOG_ERROR, "Attempt to Ref a null pointer\n");
- return p;
-}
-
-static struct rtmp_pvt *streamgroup_unref(struct rtmp_pvt *p, char *tag)
-{
- if (p)
- ao2_ref(p, -1);
- return NULL;
-}
-
-static struct rtmp_message *rtmpmessage_ref(struct rtmp_message *p, char *tag)
-{
- if (p)
- ao2_ref(p, 1);
- else
- ast_log(LOG_ERROR, "Attempt to Ref a null pointer\n");
- return p;
-}
-
-static struct rtmp_rtmpmessage *rtmpmessage_unref(struct rtmp_message *p, char *tag)
-{
- if (p)
- ao2_ref(p, -1);
- return NULL;
-}
-#endif
-
-/**
- * A maximum of 64 simultaneous channels can be supported. Therefore, as each
- * stream reserves 5 channels, we have a maximum of 11 concurrent streams
- * (channel range 0-3 is reserved).
- *
- * The following code is taken from the Red5 server sources. We show it here
- * to expose how the Red5 server relates stream and channel identifiers
- * together.
- * In function getStreamIdForChannel from file RTMPConnection.java :
- * return ((channelId - 4) / 5) + 1;
- *
- * Function createOutputStream from file RTMPConnection.java :
- * public OutputStream createOutputStream(int streamId) {
- * int channelId = (4 + ((streamId - 1) * 5));
- * final Channel data = getChannel(channelId++);
- * final Channel video = getChannel(channelId++);
- * final Channel audio = getChannel(channelId++);
- * // final Channel unknown = getChannel(channelId++);
- * // final Channel ctrl = getChannel(channelId++);
- * return new OutputStream(video, audio, data);
- * }
- */
-ast_mutex_t streamslock;
-static struct rtmp_channel* streams[RTMP_MAX_CHANNELS];
-unsigned int outgoing_chunksize = RTMP_CHUNK_SIZE;
-unsigned int incoming_chunksize = RTMP_CHUNK_SIZE;
-
static struct ast_channel *rtmp_request(const char *type, format_t format, const struct ast_channel *requestor, void *data, int *cause);
static int rtmp_call(struct ast_channel *ast, char *dest, int timeout);
-static void rtmp_destroy_fn(void *p);
static void rtmp_destroy(struct rtmp_pvt *p);
-static void rtmpmessage_destroy_fn(void *p);
-static void rtmpmessage_destroy(struct rtmp_message *rtmp);
static int rtmp_hangup(struct ast_channel *ast);
static struct ast_frame *rtmp_read(struct ast_channel *ast);
static int rtmp_write(struct ast_channel *ast, struct ast_frame *frame);
static enum ast_bridge_result rtmp_bridge(struct ast_channel *c0, struct ast_channel *c1, int flags, struct ast_frame **fo, struct ast_channel **rc, int timeoutms);
-static void* rtmp_start(void *data);
-static struct rtmp_pvt* find_streamgroup(struct rtmp_message* rtmp);
-static int check_handshake_reply (void *buffer, size_t size);
-static int rtmp_send_connect(struct rtmp_client *client, char *handshake);
-static int rtmp_send_pong(struct rtmp_client *client, struct rtmp_message *rtmp);
-static int rtmp_send_chunksize(struct rtmp_client *client, uint32_t newchunksize);
-static int rtmp_send_buffertime(struct rtmp_client *client, uint32_t streamid);
-static int rtmp_send_createstream(struct rtmp_client *client, double streamid);
-static int rtmp_send_closestream(struct rtmp_client *client, double streamid);
-static int rtmp_send_play(struct rtmp_client *client, uint32_t streamid, char *name);
-static int rtmp_send_publish(struct rtmp_client *client, uint32_t streamid, char *name);
-static int rtmp_send_audio(struct rtmp_client *client, struct rtmp_pvt *p, struct ast_frame *frame);
-static int amf_add_bobject(struct amf_object *object, uint8_t type, char *property, void *value);
-static int amf_destroy_object(struct amf_object *object);
-static char* rtmp_build_invoke(struct rtmp_message *rtmp, char *method, double connectionid, struct amf_object *amf, char *options, void *boolean, char *newoptions);
-static char* rtmp_build_audio(struct rtmp_message *rtmp, void* samples, int datalen);
-static int rtmp_set_header(char *header, struct rtmp_message *rtmp, int hdrlen);
-static int rtmp_set_boolean(void *message, void *value);
-static int rtmp_set_property(void *message, char *string);
-static int rtmp_set_string(void *message, char *string, size_t length);
-static int rtmp_set_number(void *message, double *number);
-static int rtmp_set_null(void *message);
-static int rtmp_set_object(void *message, struct amf_object *amf);
-static int amf_numberlen(double *number);
-static int amf_booleanlen(void *boolean);
-static int amf_strlen(char *string);
-static int amf_objlen(struct amf_object *object);
-static int rtmp_send_message(struct rtmp_client *client, char *prefix, char *message, size_t bodysize);
-static int rtmp_set_outgoing_channelinfo(struct rtmp_message *rtmp, uint8_t next_hdrlen);
-static int rtmp_set_incoming_channelinfo(void *buffer, int hdrlen, int channelid);
-static int rtmp_get_current_hdrlen(uint8_t channelid);
-static int rtmp_get_current_timestamp(uint8_t channelid);
-static int rtmp_get_current_bodylen(uint8_t channelid);
-static int rtmp_get_current_type(uint8_t channelid);
-static int rtmp_get_current_streamid(uint8_t channelid);
-static int rtmp_get_streamid(uint8_t channelid);
-static int rtmp_get_starting_channelid(uint8_t streamid);
-static int rtmp_get_header_length(char *header);
-static int rtmp_get_channelid(char *header);
-static int rtmp_get_bodylen(char *header, struct rtmp_message *rtmp, int direction);
-static int rtmp_parse_header(struct rtmp_message *rtmp, void *buffer);
-static int rtmp_handle_system_message(struct rtmp_client *client, struct rtmp_message *rtmp);
-static int rtmp_handle_connection_message(struct rtmp_client *client, struct rtmp_message *rtmp);
-static int rtmp_handle_audio_packet(struct rtmp_client *client, struct rtmp_message *rtmp);
-static int amf_parse_reply(double *result, char *level, char *code, char *description, char *amf, size_t len);
-static int amf_get_type(char *buf);
-static int amf_get_string(char *string, void* buffer, size_t length);
-static int amf_get_number(double *number, void *amf);
-static int rtmp_new_streamid(int range);
-static int isfree(int streamid, int range);
-static int activate_channels(int channelid, int range);
-static int desactivate_channels(int channelid, int range);
+static void *rtmp_readstream(void *data);
+static int rtmp_send_audio(struct rtmp_pvt *p, struct ast_frame *frame);
+static int rtmp_handle_apacket(struct rtmp_pvt *p, RTMPPacket *packet);
static const struct ast_channel_tech rtmp_tech = {
.type = type,
@@ -306,87 +132,66 @@
.bridge = rtmp_bridge,
};
-/*!
- * \note Use the stream id
- */
-static int streamgroups_hash_cb(const void *obj, const int flags) {
- const struct rtmp_pvt *pvt = obj;
-
- ast_debug(7, "pvt->streamid = %d\n", pvt->streamid);
- return pvt->streamid;
-}
-
-/*!
- * \note Use the stream id
- */
-static int streamgroups_cmp_cb(void *obj, void *arg, int flags) {
- int res = 0;
- struct rtmp_pvt *pvt = obj, *pvt2 = arg;
-
- ast_debug(7, "pvt->streamid = %d - pvt2->streamid = %d - pvt->numstreams = %d\n", pvt->streamid, pvt2->streamid, pvt->numstreams);
- /* match if streamid <= id < streamid + range */
- if (pvt2->streamid >= pvt->streamid && pvt2->streamid < pvt->streamid + pvt->numstreams) {
- res = CMP_MATCH | CMP_STOP;
- }
-
- return res;
-}
-
-/*!
- * \note Use the channel id
- */
-static int rtmpmessages_hash_cb(const void *obj, const int flags) {
- const struct rtmp_message *pvt = obj;
-
- ast_debug(7, "pvt->channelid = %d\n", pvt->channelid);
- return pvt->channelid;
-}
-
-/*!
- * \note Use the stream id
- */
-static int rtmpmessages_cmp_cb(void *obj, void *arg, int flags) {
- int res = 0;
- struct rtmp_message *pvt = obj, *pvt2 = arg;
-
- if (pvt2->channelid == pvt->channelid) {
- res = CMP_MATCH | CMP_STOP;
- }
-
- return res;
-}
-
-static int rtmp_call(struct ast_channel *ast, char *dest, int timeout)
-{
+static int rtmp_call(struct ast_channel *ast, char *dest, int timeout) {
struct rtmp_pvt *p;
- int res = -1;
- int i;
+ char tcUrlin[250];
+ char tcUrlout[250];
p = ast->tech_pvt;
if (!p) {
ast_debug(3, "tech_pvt is NULL\n");
- } else {
- ast_debug(3, "p->owner->name : \%s\n", p->owner->name);
+ return -1;
}
if ((ast->_state != AST_STATE_DOWN) && (ast->_state != AST_STATE_RESERVED)) {
ast_log(LOG_WARNING, "rtmp_call called on %s, neither down nor reserved\n", ast->name);
return -1;
}
- /* When we call, it just works, really, there's no destination... Just
- ring the phone and wait for someone to answer */
+
ast_debug(3, "Calling %s on %s\n", dest, ast->name);
- for (i = 0; i < p->numstreams; i++) {
- /* create at least 2 streams for publish/play */
- res = rtmp_send_createstream(client, (double)p->streamid + i);
- if (!res) {
- ast_log(LOG_WARNING, "Could not create RTMP stream\n");
- rtmp_hangup(ast);
- return -1;
- }
- ast_verbose("Sending createStream request for stream with id %f\n", (double)p->streamid + i);
- }
+ /* setup the inbound connection and associated stream */
+ snprintf(tcUrlin, sizeof(tcUrlin), "rtmp://%s:%d/%s/%s", rtmpserverstr, port, application, p->readstream);
+ RTMP_SetupURL(p->rtmpin, tcUrlin);
+ p->rtmpin->Link.lFlags |= RTMP_LF_LIVE;
+
+ if (!RTMP_Connect(p->rtmpin, NULL)) {
+ ast_log(LOG_ERROR, "Could not connect to server.\n");
+ rtmp_hangup(ast);
+ return -1;
+ }
+
+ if (!RTMP_ConnectStream(p->rtmpin, 0)) {
+ ast_log(LOG_ERROR, "Could not establish stream.\n");
+ rtmp_hangup(ast);
+ return -1;
+ }
+
+ /* now setup the outbound connection and associated stream and make
+ * sure to enable publishing */
+ snprintf(tcUrlout, sizeof(tcUrlout), "rtmp://%s:%d/%s/%s", rtmpserverstr, port, application, p->writestream);
+ RTMP_SetupURL(p->rtmpout, tcUrlout);
+ p->rtmpout->Link.lFlags |= RTMP_LF_LIVE;
+
+ RTMP_EnableWrite(p->rtmpout);
+
+ if (!RTMP_Connect(p->rtmpout, NULL)) {
+ ast_log(LOG_ERROR, "Could not connect to server.\n");
+ rtmp_hangup(ast);
+ return -1;
+ }
+
+ if (!RTMP_ConnectStream(p->rtmpout, 0)) {
+ ast_log(LOG_ERROR, "Could not establish stream.\n");
+ rtmp_hangup(ast);
+ return -1;
+ }
+
+ ast_pthread_create_background(&p->thread, NULL, rtmp_readstream, p);
+
+ /* the RTMP stream is connected */
+ ast_queue_control(p->owner, AST_CONTROL_ANSWER);
+
return 0;
}
@@ -408,19 +213,17 @@
}
av_free(p->encoding_context);
av_free(p->decoding_context);
-}
-
-static void rtmpmessage_destroy_fn(void *p) {
- rtmpmessage_destroy(p);
-}
-
-static void rtmpmessage_destroy(struct rtmp_message *rtmp) {
- ast_free(rtmp->body);
-}
-
-/** \brief Allocate a new RTMP stream
- * A new RTMP stream consists of 5 RTMP channels
- */
+
+ pthread_cancel(p->thread);
+
+ RTMP_Close(p->rtmpin);
+ RTMP_Free(p->rtmpin);
+
+ RTMP_Close(p->rtmpout);
+ RTMP_Free(p->rtmpout);
+}
+
+/** \brief Allocate a new RTMP stream */
static struct rtmp_pvt *rtmp_alloc(char *writestream, char *readstream, char *readnum) {
struct rtmp_pvt *p;
int rnum = 0;
@@ -431,31 +234,16 @@
rnum = atoi(readnum);
}
- if (!(p = ao2_t_alloc(sizeof(*p), rtmp_destroy_fn, "allocate an streamgroup(pvt) struct")))
+ if (!(p = ao2_t_alloc(sizeof(*p), rtmp_destroy_fn, "allocate a rtmp_pvt struct"))) {
return NULL;
-
- /* Each rtmp_pvt struct spans over a block of at least 2 RTMP streams :
- * the first stream identifier is reserved for publication, the
- * following stream identifiers are reserved for stream playback.
- * streamid identifies the first stream identifier for this span,
- * numstreams gives the total number of streams.
- */
- p->streamid = rtmp_new_streamid(1 + rnum);
-
- if (!p->streamid) {
- ast_log(LOG_WARNING, "Could not find a block of %d streams for this rtmp_pvt\n", 1 + rnum);
- streamgroup_unref(p, "Released a reference");
- return NULL;
- }
-
- p->numstreams = 1 + rnum;
- p->readstream_index = 0;
+ }
+
p->encoder = NULL;
p->decoder = NULL;
p->encoding_context = NULL;
p->decoding_context = NULL;
- p->rtmpinputrate = 11000;
- p->astinputrate = 8000;
+ p->rtmpinputrate = rtmpinputrate;
+ p->astinputrate = astinputrate;
/* the outputrate value of this context matches with the sampling
* rate of the RTMP packets that come in to Asterisk. On the other
@@ -472,35 +260,33 @@
* int linear, double cutoff)
*/
p->tortmp_resample_context = av_audio_resample_init(
- 1, 1,
- p->rtmpinputrate, p->astinputrate,
- SAMPLE_FMT_S16, SAMPLE_FMT_S16,
- 16, 10, 1, 0.8);
+ 1, 1,
+ p->rtmpinputrate, p->astinputrate,
+ SAMPLE_FMT_S16, SAMPLE_FMT_S16,
+ 16, 10, 1, 0.8);
/* see the comment above */
p->fromrtmp_resample_context = av_audio_resample_init(
- 1, 1,
- p->astinputrate, p->rtmpinputrate,
- SAMPLE_FMT_S16, SAMPLE_FMT_S16,
- 16, 10, 1, 0.8);
+ 1, 1,
+ p->astinputrate, p->rtmpinputrate,
+ SAMPLE_FMT_S16, SAMPLE_FMT_S16,
+ 16, 10, 1, 0.8);
strncpy(p->readstream, readstream, AST_MAX_EXTENSION);
strncpy(p->writestream, writestream, AST_MAX_EXTENSION);
- /* add to active RTMP streams list */
- ao2_t_link(streamgroups, p, "link pvt into RTMP streams table");
+ p->rtmpin = RTMP_Alloc();
+ RTMP_Init(p->rtmpin);
+
+ p->rtmpout = RTMP_Alloc();
+ RTMP_Init(p->rtmpout);
return p;
}
static int rtmp_hangup(struct ast_channel *ast) {
struct rtmp_pvt *p;
- int i;
p = ast->tech_pvt;
-
- for (i = 0; i < p->numstreams; i++) {
- rtmp_send_closestream(client, p->streamid + i);
- }
ast_debug(3, "rtmp_hangup(%s)\n", ast->name);
if (!ast->tech_pvt) {
@@ -508,13 +294,7 @@
return 0;
}
- desactivate_channels(rtmp_get_starting_channelid(p->streamid), p->numstreams);
-
- streamgroup_ref(p, "Let's bump the count in the unlink so it doesn't accidentally become dead before we are done");
- ast_debug(3, "Deleting rtmp_pvt message from list\n");
- ao2_t_unlink(streamgroups, p, "unlinking RTMP streams group via ao2_unlink");
- streamgroup_unref(p, "Dereferecing RTMP streams group after it has been unlinked");
- streamgroup_unref(p, "Closing stream");
+ rtmp_destroy(p);
ast->tech_pvt = NULL;
ast_setstate(ast, AST_STATE_DOWN);
@@ -523,11 +303,10 @@
static struct ast_frame *rtmp_read(struct ast_channel *ast) {
struct rtmp_pvt *p = ast->tech_pvt;
+ static char buf[4096];
int res;
- char *buf = NULL;
static struct ast_frame f;
- buf = ast_malloc(4096);
if (!buf) {
return NULL;
}
@@ -557,7 +336,6 @@
ast_debug(7, "Read %d bytes as a frame on %s\n", res, ast->name);
- ast_free(buf);
return &f;
}
@@ -579,35 +357,34 @@
}
if (ast->_state != AST_STATE_UP) {
- /* Don't try tos end audio on-hook */
return 0;
}
if (frame->frametype == AST_FRAME_VOICE) {
- res = rtmp_send_audio(client, p, frame);
- }
-
- return 0;
+ res = rtmp_send_audio(p, frame);
+ }
+
+ return res;
}
static enum ast_bridge_result rtmp_bridge(struct ast_channel *c0, struct ast_channel *c1, int flags, struct ast_frame **fo, struct ast_channel **rc, int timeoutms) {
return 0;
}
-static struct ast_channel *rtmp_new(struct rtmp_pvt *i, int state, const char *linkedid)
+static struct ast_channel *rtmp_new(struct rtmp_pvt *p, int state, const char *linkedid)
{
struct ast_channel *tmp;
- tmp = ast_channel_alloc(1, state, 0, 0, "", linkedid, "s", context, 0, "RTMP/%d", i->streamid);
+ tmp = ast_channel_alloc(1, state, 0, 0, "", linkedid, "s", context, 0, "RTMP/%s-%s-%04lx", p->readstream, p->writestream, ast_random() & 0xffff);
if (!tmp) {
ast_log(LOG_WARNING, "Unable to allocate channel structure\n");
return NULL;
}
- if (pipe(i->pipe) < 0) {
+ if (pipe(p->pipe) < 0) {
ast_log(LOG_ERROR, "Pipe failed\n");
}
- ast_channel_set_fd(tmp, 0, i->pipe[0]);
+ ast_channel_set_fd(tmp, 0, p->pipe[0]);
tmp->tech = &rtmp_tech;
tmp->nativeformats = prefformat;
@@ -615,31 +392,31 @@
tmp->writeformat = prefformat;
if (state == AST_STATE_RING)
tmp->rings = 1;
- tmp->tech_pvt = i;
+ tmp->tech_pvt = p;
ast_copy_string(tmp->context, context, sizeof(tmp->context));
ast_copy_string(tmp->exten, "s", sizeof(tmp->exten));
ast_string_field_set(tmp, language, "");
- i->owner = tmp;
- i->encoder = avcodec_find_encoder(CODEC_ID_PCM_S16LE);
- if (!i->encoder) {
+ p->owner = tmp;
+ p->encoder = avcodec_find_encoder(CODEC_ID_PCM_S16LE);
+ if (!p->encoder) {
ast_debug(3, "Codec not found\n");
ast_hangup(tmp);
}
- i->encoding_context = avcodec_alloc_context2(CODEC_ID_ADPCM_SWF);
- i->encoding_context->channels = 1;
- i->encoding_context->sample_rate = 11000;
- if (avcodec_open(i->encoding_context, i->encoder) < 0) {
+ p->encoding_context = avcodec_alloc_context();
+ p->encoding_context->channels = 1;
+ p->encoding_context->sample_rate = 11000;
+ if (avcodec_open(p->encoding_context, p->encoder) < 0) {
ast_debug(3, "Could not open codec\n");
ast_hangup(tmp);
}
- i->decoder = avcodec_find_decoder(CODEC_ID_NELLYMOSER);
- if (!i->decoder) {
+ p->decoder = avcodec_find_decoder(CODEC_ID_NELLYMOSER);
+ if (!p->decoder) {
ast_debug(3, "Codec not found\n");
ast_hangup(tmp);
}
- i->decoding_context = avcodec_alloc_context2(CODEC_ID_NELLYMOSER);
- if (avcodec_open(i->decoding_context, i->decoder) < 0) {
+ p->decoding_context = avcodec_alloc_context();
+ if (avcodec_open(p->decoding_context, p->decoder) < 0) {
ast_debug(3, "Could not open codec\n");
ast_hangup(tmp);
}
@@ -704,1866 +481,71 @@
return tmp;
}
-static void *rtmp_start(void *data) {
- int res = -1;
- char *handshake;
- char *handshake_server; /* received from server */
- char *buffer;
- struct ao2_container *rtmpmessages = NULL;
- struct ao2_iterator aux;
- struct rtmp_message *rtmp = NULL;
-
- buffer = ast_malloc(RTMP_RECV_BUFSIZE);
- if (!buffer) {
- return NULL;
- }
- handshake = ast_malloc(RTMP_BLOCK_SIZE + 1);
- if (!handshake) {
- return NULL;
- }
- handshake_server = ast_malloc(RTMP_BLOCK_SIZE);
- if (!handshake_server) {
- return NULL;
- }
- /* initialize the RTMP messages container */
- rtmpmessages = ao2_t_container_alloc(hash_rtmpmessages_size, rtmpmessages_hash_cb, rtmpmessages_cmp_cb, "allocate RTMP messages");
-
- /* set the first byte to 0x03, fill the rest with zeros */
- memset(handshake, '\0', RTMP_BLOCK_SIZE + 1);
- handshake[0] = '\3';
-
-
- ast_debug(3, "Starting thread\n");
- client->state = RTMP_CONNECTING;
- client->fd = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
- if (!client->fd) {
- ast_log(LOG_ERROR, "Unable to build socket\n");
- return NULL;
- }
- if (connect(client->fd, (struct sockaddr *) &rtmpserver, sizeof(rtmpserver)) < 0) {
- ast_log(LOG_ERROR, "Unable to connect to server\n");
- return NULL;
- }
-
- /* start handshake */
- if (send(client->fd, handshake, RTMP_BLOCK_SIZE + 1, 0) != RTMP_BLOCK_SIZE + 1) {
- ast_log(LOG_ERROR, "Could not initiate handshake\n");
- return NULL;
- }
-
- /* wait for the server to reply, check answer */
- res = recv(client->fd, buffer, 2*RTMP_BLOCK_SIZE + 1, MSG_WAITALL);
- memcpy(handshake_server, buffer + 1, RTMP_BLOCK_SIZE);
- client->state = check_handshake_reply(buffer, res);
-
- /* mark main RTMP channels (0 to 3) as active */
- activate_channels(0, 0);
-
- /* send connect message and wait for reply */
- rtmp_send_connect(client, handshake_server);
- client->state = RTMP_CONNECTED;
-
- while(client->fd && res) {
- int buflen = 0;
- int channelid = 0;
- struct rtmp_message mtmp;
- int newmessage = 0;
+static void *rtmp_readstream(void *data) {
+ struct rtmp_pvt *p = data;
+ RTMPPacket packet = { 0 };
+
+ while (p->rtmpin && RTMP_IsConnected(p->rtmpin)) {
+ RTMP_GetNextMediaPacket(p->rtmpin, &packet);
- memset(buffer, '\0', RTMP_RECV_BUFSIZE);
-
- /* receive first byte to get header length and channel id */
- res = recv(client->fd, buffer, 1, MSG_WAITALL);
- if (!res) {
- ast_log(LOG_WARNING, "RTMP socket closed\n");
- break;
- }
-
- ast_debug(7, "--------- New RTMP message received --------\n");
- ast_debug(7, "res = %d\n", res);
- channelid = rtmp_get_channelid(buffer);
- mtmp.channelid = channelid;
- rtmp = ao2_t_find(rtmpmessages, &mtmp, OBJ_POINTER, "ao2 find in rtmpmessages");
- if (!rtmp) {
- newmessage = 1;
- /* need to build a new message and insert it in our list */
- if (!(rtmp = ao2_t_alloc(sizeof(*rtmp), rtmpmessage_destroy_fn, "allocate RTMP message struct"))) {
- return NULL;
- }
- rtmp->body = ast_malloc(RTMP_MAX_BODYSIZE);
- if (!rtmp->body) {
- return NULL;
- }
- memset(rtmp->body, '\0', RTMP_MAX_BODYSIZE);
- }
-
- rtmp->channelid = rtmp_get_channelid(buffer);
- rtmp->hdrlen = rtmp_get_header_length(buffer);
-
- /* retrieve the remaining header bytes */
- if (rtmp->hdrlen > 1) {
- ast_debug(7, "hdrlen = %d\n", rtmp->hdrlen);
- //res = recv(client->fd, buffer, rtmp->hdrlen - 1, 0);
- res = recv(client->fd, buffer, rtmp->hdrlen - 1, MSG_WAITALL);
- ast_debug(7, "res = %d\n", res);
- rtmp_parse_header(rtmp, buffer);
- rtmp_set_incoming_channelinfo(buffer, rtmp->hdrlen, rtmp->channelid);
- }
- rtmp->bodysize = rtmp_get_bodylen(buffer, rtmp, RTMP_INCOMING);
- ast_debug(7, "rtmp->bodysize = %d\n", rtmp->bodysize);
-
- if (rtmp->bodysize <= incoming_chunksize) {
- buflen = rtmp->bodysize;
- } else {
- if (rtmp->bodysize - rtmp->bytesread <= incoming_chunksize) {
- buflen = rtmp->bodysize - rtmp->bytesread;
- } else {
- buflen = incoming_chunksize;
- }
- }
-
- if (!rtmp->bodysize) {
- /* message has no body */
- if (!newmessage) {
- rtmpmessage_ref(rtmp, "Let's bump the count in the unlink so it doesn't accidentally become dead before we are done");
-
- ast_debug(7, "Deleting RTMP message from list\n");
- ao2_t_unlink(rtmpmessages, rtmp, "unlinking RTMP message via ao2_unlink");
- rtmpmessage_unref(rtmp, "Dereferecing RTMP message after it has been unlinked");
- }
- rtmpmessage_unref(rtmp, "release reference on RTMP message, should be destroyed now");
+ if (!packet.m_nBodySize) {
+ /* ignore zero length media packets */
continue;
}
- /* retrieve the body parts */
- res = recv(client->fd, buffer, buflen, 0);
- //res = recv(client->fd, buffer, buflen, MSG_WAITALL);
- ast_debug(7, "res = %d\n", res);
-
- memcpy(rtmp->body + rtmp->bytesread, buffer, buflen);
- rtmp->bytesread += buflen;
- ast_debug(7, "rtmp->bytesread = %d\n", rtmp->bytesread);
-
- if (rtmp->bytesread < rtmp->bodysize) {
- /* message has been partially retrieved, release
- * reference and link it to the messages list if it
- * was not found */
- if (newmessage) {
- ast_debug(5, "Inserted new RTMP message into list\n");
- ao2_t_link(rtmpmessages, rtmp, "link into RTMP messages table");
- rtmpmessage_unref(rtmp, "Released a reference (rtmp_message)");
- } else {
- rtmpmessage_unref(rtmp, "Released a reference (rtmp_message)");
-
- }
- continue;
- }
-
- if (!streams[rtmp->channelid]) {
- ast_log(LOG_WARNING, "Ignoring message received on inactive RTMP channel %d\n", rtmp->channelid);
- continue;
- }
-
- /* message has been completely retrieved, process it */
-
- if (rtmp->channelid < 4) {
- /* handle system messages here */
- switch (rtmp->channelid) {
- case 0:
- break;
- case 1:
- break;
- case RTMP_CHANNEL_SYSTEM:
- rtmp_handle_system_message(client, rtmp);
- break;
- case RTMP_CHANNEL_CONNECT:
- rtmp_handle_connection_message(client, rtmp);
- break;
- }
- /* release the reference on this message, which should be
- * destroyed by Asterisk */
- if (!newmessage) {
- rtmpmessage_ref(rtmp, "Let's bump the count in the unlink so it doesn't accidentally become dead before we are done");
-
- ast_debug(5, "Deleting RTMP message from list\n");
- ao2_t_unlink(rtmpmessages, rtmp, "unlinking RTMP message via ao2_unlink");
- rtmpmessage_unref(rtmp, "Dereferecing RTMP message after it has been unlinked");
- }
- rtmpmessage_unref(rtmp, "release reference on RTMP message, should be destroyed now");
- continue;
- }
-
- ast_debug(5, "rtmp->channelid = %d\n", rtmp->channelid);
-
- switch ((rtmp->channelid + 1) % RTMP_STREAM_CHANNEL_RANGE) {
-
- case RTMP_CHANNEL_DATA:
- ast_debug(5, "Received DATA message for channel with id %d\n", rtmp->channelid);
- /* TODO : properly handle replies to createStream here */
- /* rtmp_handle_connection_message(client, rtmp); */
+ switch (packet.m_packetType) {
+ case RTMP_PACKET_TYPE_AUDIO:
+ ast_debug(7, "Received audio packet.\n");
+ ast_debug(7, "RTMP PACKET: packet type: 0x%02x. channel: 0x%02x. info 1: %d info 2: %d. Body size: %u. body: 0x%02x\n",
+ packet.m_packetType, packet.m_nChannel, packet.m_nTimeStamp, packet.m_nInfoField2,
+ packet.m_nBodySize, packet.m_body ? (unsigned char)packet.m_body[0] : 0);
+ rtmp_handle_apacket(p, &packet);
break;
- case RTMP_CHANNEL_VIDEO:
- ast_debug(5, "Received VIDEO message for channel with id %d\n", rtmp->channelid);
+ case RTMP_PACKET_TYPE_VIDEO:
+ ast_debug(7, "Received video packet.\n");
+ ast_debug(7, "RTMP PACKET: packet type: 0x%02x. channel: 0x%02x. info 1: %d info 2: %d. Body size: %u. body: 0x%02x\n",
+ packet.m_packetType, packet.m_nChannel, packet.m_nTimeStamp, packet.m_nInfoField2,
+ packet.m_nBodySize, packet.m_body ? (unsigned char)packet.m_body[0] : 0);
break;
- case RTMP_CHANNEL_AUDIO:
- ast_debug(5, "Received AUDIO message for channel with id %d\n", rtmp->channelid);
- rtmp_handle_audio_packet(client, rtmp);
- break;
- case RTMP_CHANNEL_UNKNOWN:
- ast_debug(5, "Received UNKNOWN message for channel with id %d\n", rtmp->channelid);
- break;
- case RTMP_CHANNEL_CONTROL:
- ast_debug(5, "Received CONTROL message for channel with id %d\n", rtmp->channelid);
+ default:
+ ast_debug(7, "Received unknown packet type : %d\n", packet.m_packetType);
+ ast_debug(7, "RTMP PACKET: packet type: 0x%02x. channel: 0x%02x. info 1: %d info 2: %d. Body size: %u. body: 0x%02x\n",
+ packet.m_packetType, packet.m_nChannel, packet.m_nTimeStamp, packet.m_nInfoField2,
+ packet.m_nBodySize, packet.m_body ? (unsigned char)packet.m_body[0] : 0);
break;
}
-
- /* release the reference on this message, which should be
- * destroyed by Asterisk */
- if (!newmessage) {
- rtmpmessage_ref(rtmp, "Let's bump the count in the unlink so it doesn't accidentally become dead before we are done");
-
- ast_debug(5, "Deleting RTMP message from list\n");
- ao2_t_unlink(rtmpmessages, rtmp, "unlinking RTMP message via ao2_unlink");
- rtmpmessage_unref(rtmp, "Dereferecing RTMP message after it has been unlinked");
- }
- rtmpmessage_unref(rtmp, "release reference on RTMP message, should be destroyed now");
- }
-
- /* Free the RTMP messages list */
- aux = ao2_iterator_init(rtmpmessages, 0);
- while ((rtmp = ao2_t_iterator_next(&aux, "iterate thru RTMP messages"))) {
- rtmpmessage_unref(rtmp, "toss RTMP message ptr from iterator_next");
- }
- ast_free(handshake);
- ast_free(handshake_server);
- ast_free(buffer);
- close(client->fd);
+ }
+
+ ast_debug(7, "Left loop.\n");
return 0;
}
-static struct rtmp_pvt* find_streamgroup(struct rtmp_message* rtmp) {
- struct rtmp_pvt *p = NULL;
- struct rtmp_pvt tmp_streamgroup;
- int streamid = -1;
-
- streamid = rtmp_get_streamid(rtmp->channelid);
- tmp_streamgroup.streamid = streamid;
-
- p = ao2_t_find(streamgroups, &tmp_streamgroup, 0, "ao2 find in streamgroups");
-
- return p;
-}
-
-static int check_handshake_reply(void *buffer, size_t size) {
+/** \brief Handle audio packets
+ *
+ * The first byte is not a media packet,
+ * it contains the following codec information :
+ * soundType (byte & 0x01) >> 0 0: mono, 1: stereo
+ * soundSize (byte & 0x02) >> 1 0: 8-bit, 1: 16-bit
+ * soundRate (byte & 0x0c) >> 2 0: 5.5 kHz, 1: 11 kHz, 2: 22 kHz, 3: 44 kHz
+ * soundFormat (byte & 0xf0) >> 4 0: Uncompressed, 1: ADPCM, 2: MP3, 5: Nellymoser 8kHz mono, 6: Nellymoser, 11: Speex
+ */
+static int rtmp_handle_apacket(struct rtmp_pvt *p, RTMPPacket *packet) {
int res = -1;
- char handshake[2*RTMP_BLOCK_SIZE + 1];
-
- /* expected reply */
- memset(&handshake, 0x00, 2*RTMP_BLOCK_SIZE + 1);
- handshake[0] = 0x03;
- handshake[4] = 0x01;
-
- if (!memcmp(handshake, buffer, 2*RTMP_BLOCK_SIZE)) {
- /* skip last byte because its value is not always the same! */
- ast_debug(3, "Handshake test passed, buffer size = %d\n", (int)size);
- res = RTMP_HANDSHAKE_OK;
- }
-
- return res;
-}
-
-static int rtmp_send_connect(struct rtmp_client *client, char *handshake) {
- int res = -1;
- struct amf_object *object = NULL;
- struct rtmp_message *rtmp = NULL;
- void *message = NULL;
- double videocodecs = 0.0;
- double audiocodecs = 615.0;
- char tcUrl[50];
-
- ast_debug(3, "In rtmp_send_connect\n");
-
- /* build URL */
- snprintf(tcUrl, sizeof(tcUrl), "rtmp://%s:%d/%s", rtmpserverstr, port, application);
-
- object = ast_calloc(1, sizeof(*object));
- if (!object) {
- ast_log(LOG_ERROR, "Memory allocation failure\n");
- return res;
- }
- rtmp = ast_calloc(1, sizeof(*rtmp));
- if (!rtmp) {
- ast_log(LOG_ERROR, "Memory allocation failure\n");
- return res;
- }
-
- /* build header */
- rtmp->hdrlen = 12;
- rtmp->type = RTMP_TYPE_INVOKE;
- rtmp->channelid = RTMP_CHANNEL_CONNECT;
- rtmp->timestamp = 0;
- rtmp->streamid = 0;
-
- object->size = 0;
-
- /* Populate the AMF object
- * In the case of the INVOKE method, the AMF object must contain
- * the following basic objects :
- * - app : the application identifier to connect to
- * - swfUul : referrer to the SWF file
- * - tcUrl : the RTMP URL
- * - flashVer : agent version
- * - audioCodecs
- * - videoCodecs
- * - page URL
- * Property names are case sensitive */
- amf_add_bobject(object, AMF_TYPE_STRING, "app", application);
- amf_add_bobject(object, AMF_TYPE_STRING, "flashVer", "LNX 9,0,31,0");
- // amf_add_bobject(object, AMF_TYPE_STRING, "swfUrl", "http://xcom.inria.fr/TestVideo/TestVideo.swf");
- amf_add_bobject(object, AMF_TYPE_STRING, "tcUrl", tcUrl);
- amf_add_bobject(object, AMF_TYPE_BOOLEAN, "fpad", AMF_BOOLEAN_FALSE);
- amf_add_bobject(object, AMF_TYPE_NUMBER, "audioCodecs", &audiocodecs);
- amf_add_bobject(object, AMF_TYPE_NUMBER, "videoCodecs", &videocodecs);
- // amf_add_bobject(object, AMF_TYPE_STRING, "pageUrl", "http://xcom.inria.fr/TestVideo/TestVideo.html");
-
- ast_debug(3, "Calling rtmp_send_invoke\n");
- message = rtmp_build_invoke(rtmp, "connect", 1.0, object, NULL, NULL, NULL);
- if (!message) {
- ast_log(LOG_ERROR, "Could not set buffer\n");
- goto safeout;
- }
-
- res = rtmp_send_message(client, handshake, message, rtmp->bodysize);
- ast_debug(3, "Sent %d bytes to server\n", res);
-
-safeout:
- amf_destroy_object(object);
- ast_free(rtmp);
- ast_free(message);
- return res;
-}
-
-/** \brief Send PONG message back to server
- * \param rtmp the received PING message
- * \note body size is the same in both directions
- *
- */
-static int rtmp_send_pong(struct rtmp_client *client, struct rtmp_message *rtmp) {
- int res = -1;
- void *message = NULL;
- void *aux = NULL;
- uint16_t pingtype = htons(RTMP_PING_TYPE_PONG);
- struct rtmp_message *pong = NULL;
- int hdrlen = 0;
- int msglen = 0;
- int current_bodylen = 0;
- int current_type = 0;
-
-
- pong = ast_calloc(1, sizeof(*pong));
- if (!pong) {
- ast_log(LOG_ERROR, "Memory allocation failure\n");
- return res;
- }
- /* populate our PONG packet */
- pong->channelid = RTMP_CHANNEL_SYSTEM;
- pong->bodysize = RTMP_PING_DEFAULTBODYSIZE;
- pong->type = RTMP_TYPE_PING;
- pong->streamid = 0;
- pong->timestamp = 0;
-
- current_bodylen = rtmp_get_current_bodylen(rtmp->channelid);
- current_type = rtmp_get_current_type(rtmp->channelid);
- rtmp_get_current_timestamp(rtmp->channelid);
- rtmp_get_current_streamid(rtmp->channelid);
-
-
- if (pong->type != current_type || pong->bodysize != current_bodylen) {
- hdrlen = 12;
- } else {
- hdrlen = 1;
- }
-
- pong->hdrlen = hdrlen;
- msglen = hdrlen + RTMP_PING_DEFAULTBODYSIZE;
-
- message = ast_calloc(1, msglen);
- if (!message) {
- res = -1;
- goto safeout;
- }
-
- aux = message;
- res = rtmp_set_header(aux, pong, hdrlen);
- if (!res) {
- ast_log(LOG_ERROR, "Error while setting header\n");
- return res;
- }
-
- aux += res;
- /* set ping type (2 bytes long) */
- memcpy(aux, &pingtype, 2);
- aux += 2;
-
- /* set timestamp (4 bytes long) */
- memcpy(aux, rtmp->body + 2, 4);
-
- res = rtmp_send_message(client, NULL, message, pong->bodysize);
-
-safeout:
- ast_free(pong);
- ast_free(message);
- return res;
-}
-
[... 2082 lines stripped ...]
More information about the asterisk-commits
mailing list