#include "flv.h"
#include "rtmp.h"
+#include "rtmpcrypt.h"
#include "rtmppkt.h"
#include "url.h"
typedef enum {
STATE_START, ///< client has not done anything yet
STATE_HANDSHAKED, ///< client has performed handshake
- STATE_RELEASING, ///< client releasing stream before publish it (for output)
STATE_FCPUBLISH, ///< client FCPublishing stream (for output)
- STATE_CONNECTING, ///< client connected to server successfully
- STATE_READY, ///< client has sent all needed commands and waits for server reply
STATE_PLAYING, ///< client has started receiving multimedia data from server
STATE_PUBLISHING, ///< client has started sending multimedia data to server (for output)
STATE_STOPPED, ///< the broadcast has been stopped
} ClientState;
+typedef struct TrackedMethod {
+ char *name;
+ int id;
+} TrackedMethod;
+
/** protocol handler context */
typedef struct RTMPContext {
const AVClass *class;
uint8_t* flv_data; ///< buffer with data for demuxer
int flv_size; ///< current buffer size
int flv_off; ///< number of bytes read from current buffer
+ int flv_nb_packets; ///< number of flv packets published
RTMPPacket out_pkt; ///< rtmp packet, created from flv a/v or metadata (for output)
uint32_t client_report_size; ///< number of bytes after which client should report to server
uint32_t bytes_read; ///< number of bytes read from server
uint8_t flv_header[11]; ///< partial incoming flv packet header
int flv_header_bytes; ///< number of initialized bytes in flv_header
int nb_invokes; ///< keeps track of invoke messages
- int create_stream_invoke; ///< invoke id for the create stream command
char* tcurl; ///< url of the target stream
char* flashver; ///< version of the flash plugin
char* swfurl; ///< url of the swf player
+ char* pageurl; ///< url of the web page
+ char* subscribe; ///< name of live stream to subscribe
int server_bw; ///< server bandwidth
int client_buffer_time; ///< client buffer time in ms
+ int flush_interval; ///< number of packets flushed in the same request (RTMPT only)
+ int encrypted; ///< use an encrypted connection (RTMPE only)
+ TrackedMethod*tracked_methods; ///< tracked methods buffer
+ int nb_tracked_methods; ///< number of tracked methods
+ int tracked_methods_size; ///< size of the tracked methods buffer
} RTMPContext;
#define PLAYER_KEY_OPEN_PART_LEN 30 ///< length of partial key used for first client digest signing
0xE6, 0x36, 0xCF, 0xEB, 0x31, 0xAE
};
+static int add_tracked_method(RTMPContext *rt, const char *name, int id)
+{
+ void *ptr;
+
+ if (rt->nb_tracked_methods + 1 > rt->tracked_methods_size) {
+ rt->tracked_methods_size = (rt->nb_tracked_methods + 1) * 2;
+ ptr = av_realloc(rt->tracked_methods,
+ rt->tracked_methods_size * sizeof(*rt->tracked_methods));
+ if (!ptr)
+ return AVERROR(ENOMEM);
+ rt->tracked_methods = ptr;
+ }
+
+ rt->tracked_methods[rt->nb_tracked_methods].name = av_strdup(name);
+ if (!rt->tracked_methods[rt->nb_tracked_methods].name)
+ return AVERROR(ENOMEM);
+ rt->tracked_methods[rt->nb_tracked_methods].id = id;
+ rt->nb_tracked_methods++;
+
+ return 0;
+}
+
+static void del_tracked_method(RTMPContext *rt, int index)
+{
+ memmove(&rt->tracked_methods[index], &rt->tracked_methods[index + 1],
+ sizeof(*rt->tracked_methods) * (rt->nb_tracked_methods - index - 1));
+ rt->nb_tracked_methods--;
+}
+
+static void free_tracked_methods(RTMPContext *rt)
+{
+ int i;
+
+ for (i = 0; i < rt->nb_tracked_methods; i ++)
+ av_free(rt->tracked_methods[i].name);
+ av_free(rt->tracked_methods);
+}
+
+static int rtmp_send_packet(RTMPContext *rt, RTMPPacket *pkt, int track)
+{
+ int ret;
+
+ if (pkt->type == RTMP_PT_INVOKE && track) {
+ GetByteContext gbc;
+ char name[128];
+ double pkt_id;
+ int len;
+
+ bytestream2_init(&gbc, pkt->data, pkt->data_size);
+ if ((ret = ff_amf_read_string(&gbc, name, sizeof(name), &len)) < 0)
+ goto fail;
+
+ if ((ret = ff_amf_read_number(&gbc, &pkt_id)) < 0)
+ goto fail;
+
+ if ((ret = add_tracked_method(rt, name, pkt_id)) < 0)
+ goto fail;
+ }
+
+ ret = ff_rtmp_packet_write(rt->stream, pkt, rt->chunk_size,
+ rt->prev_pkt[1]);
+fail:
+ ff_rtmp_packet_destroy(pkt);
+ return ret;
+}
+
static int rtmp_write_amf_data(URLContext *s, char *param, uint8_t **p)
{
char *field, *value;
ff_amf_write_number(&p, 252.0);
ff_amf_write_field_name(&p, "videoFunction");
ff_amf_write_number(&p, 1.0);
+
+ if (rt->pageurl) {
+ ff_amf_write_field_name(&p, "pageUrl");
+ ff_amf_write_string(&p, rt->pageurl);
+ }
}
ff_amf_write_object_end(&p);
pkt.data_size = p - pkt.data;
- ret = ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size,
- rt->prev_pkt[1]);
- ff_rtmp_packet_destroy(&pkt);
-
- return ret;
+ return rtmp_send_packet(rt, &pkt, 1);
}
/**
ff_amf_write_null(&p);
ff_amf_write_string(&p, rt->playpath);
- ret = ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size,
- rt->prev_pkt[1]);
- ff_rtmp_packet_destroy(&pkt);
-
- return ret;
+ return rtmp_send_packet(rt, &pkt, 0);
}
/**
ff_amf_write_null(&p);
ff_amf_write_string(&p, rt->playpath);
- ret = ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size,
- rt->prev_pkt[1]);
- ff_rtmp_packet_destroy(&pkt);
-
- return ret;
+ return rtmp_send_packet(rt, &pkt, 0);
}
/**
ff_amf_write_null(&p);
ff_amf_write_string(&p, rt->playpath);
- ret = ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size,
- rt->prev_pkt[1]);
- ff_rtmp_packet_destroy(&pkt);
-
- return ret;
+ return rtmp_send_packet(rt, &pkt, 0);
}
/**
ff_amf_write_string(&p, "createStream");
ff_amf_write_number(&p, ++rt->nb_invokes);
ff_amf_write_null(&p);
- rt->create_stream_invoke = rt->nb_invokes;
- ret = ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size,
- rt->prev_pkt[1]);
- ff_rtmp_packet_destroy(&pkt);
-
- return ret;
+ return rtmp_send_packet(rt, &pkt, 1);
}
ff_amf_write_null(&p);
ff_amf_write_number(&p, rt->main_channel_id);
- ret = ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size,
- rt->prev_pkt[1]);
- ff_rtmp_packet_destroy(&pkt);
-
- return ret;
+ return rtmp_send_packet(rt, &pkt, 0);
}
/**
bytestream_put_be32(&p, rt->main_channel_id);
bytestream_put_be32(&p, rt->client_buffer_time);
- ret = ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size,
- rt->prev_pkt[1]);
- ff_rtmp_packet_destroy(&pkt);
-
- return ret;
+ return rtmp_send_packet(rt, &pkt, 0);
}
/**
ff_amf_write_string(&p, rt->playpath);
ff_amf_write_number(&p, rt->live);
- ret = ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size,
- rt->prev_pkt[1]);
- ff_rtmp_packet_destroy(&pkt);
-
- return ret;
+ return rtmp_send_packet(rt, &pkt, 1);
}
/**
ff_amf_write_string(&p, rt->playpath);
ff_amf_write_string(&p, "live");
- ret = ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size,
- rt->prev_pkt[1]);
- ff_rtmp_packet_destroy(&pkt);
-
- return ret;
+ return rtmp_send_packet(rt, &pkt, 1);
}
/**
uint8_t *p;
int ret;
+ if (ppkt->data_size < 6) {
+ av_log(s, AV_LOG_ERROR, "Too short ping packet (%d)\n",
+ ppkt->data_size);
+ return AVERROR_INVALIDDATA;
+ }
+
if ((ret = ff_rtmp_packet_create(&pkt, RTMP_NETWORK_CHANNEL, RTMP_PT_PING,
ppkt->timestamp + 1, 6)) < 0)
return ret;
p = pkt.data;
bytestream_put_be16(&p, 7);
bytestream_put_be32(&p, AV_RB32(ppkt->data+2));
- ret = ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size,
- rt->prev_pkt[1]);
- ff_rtmp_packet_destroy(&pkt);
- return ret;
+ return rtmp_send_packet(rt, &pkt, 0);
}
/**
p = pkt.data;
bytestream_put_be32(&p, rt->server_bw);
- ret = ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size,
- rt->prev_pkt[1]);
- ff_rtmp_packet_destroy(&pkt);
- return ret;
+ return rtmp_send_packet(rt, &pkt, 0);
}
/**
p = pkt.data;
ff_amf_write_string(&p, "_checkbw");
- ff_amf_write_number(&p, ++rt->nb_invokes);
+ ff_amf_write_number(&p, RTMP_NOTIFICATION);
ff_amf_write_null(&p);
- ret = ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size,
- rt->prev_pkt[1]);
- ff_rtmp_packet_destroy(&pkt);
-
- return ret;
+ return rtmp_send_packet(rt, &pkt, 0);
}
/**
p = pkt.data;
bytestream_put_be32(&p, rt->bytes_read);
- ret = ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size,
- rt->prev_pkt[1]);
- ff_rtmp_packet_destroy(&pkt);
- return ret;
+ return rtmp_send_packet(rt, &pkt, 0);
}
-//TODO: Move HMAC code somewhere. Eventually.
-#define HMAC_IPAD_VAL 0x36
-#define HMAC_OPAD_VAL 0x5C
+static int gen_fcsubscribe_stream(URLContext *s, RTMPContext *rt,
+ const char *subscribe)
+{
+ RTMPPacket pkt;
+ uint8_t *p;
+ int ret;
-/**
- * Calculate HMAC-SHA2 digest for RTMP handshake packets.
- *
- * @param src input buffer
- * @param len input buffer length (should be 1536)
- * @param gap offset in buffer where 32 bytes should not be taken into account
- * when calculating digest (since it will be used to store that digest)
- * @param key digest key
- * @param keylen digest key length
- * @param dst buffer where calculated digest will be stored (32 bytes)
- */
-static int rtmp_calc_digest(const uint8_t *src, int len, int gap,
- const uint8_t *key, int keylen, uint8_t *dst)
+ if ((ret = ff_rtmp_packet_create(&pkt, RTMP_SYSTEM_CHANNEL, RTMP_PT_INVOKE,
+ 0, 27 + strlen(subscribe))) < 0)
+ return ret;
+
+ p = pkt.data;
+ ff_amf_write_string(&p, "FCSubscribe");
+ ff_amf_write_number(&p, ++rt->nb_invokes);
+ ff_amf_write_null(&p);
+ ff_amf_write_string(&p, subscribe);
+
+ return rtmp_send_packet(rt, &pkt, 1);
+}
+
+int ff_rtmp_calc_digest(const uint8_t *src, int len, int gap,
+ const uint8_t *key, int keylen, uint8_t *dst)
{
struct AVSHA *sha;
uint8_t hmac_buf[64+32] = {0};
return 0;
}
+int ff_rtmp_calc_digest_pos(const uint8_t *buf, int off, int mod_val,
+ int add_val)
+{
+ int i, digest_pos = 0;
+
+ for (i = 0; i < 4; i++)
+ digest_pos += buf[i + off];
+ digest_pos = digest_pos % mod_val + add_val;
+
+ return digest_pos;
+}
+
/**
* Put HMAC-SHA2 digest of packet data (except for the bytes where this digest
* will be stored) into that packet.
*
* @param buf handshake data (1536 bytes)
+ * @param encrypted use an encrypted connection (RTMPE)
* @return offset to the digest inside input data
*/
-static int rtmp_handshake_imprint_with_digest(uint8_t *buf)
+static int rtmp_handshake_imprint_with_digest(uint8_t *buf, int encrypted)
{
- int i, digest_pos = 0;
- int ret;
+ int ret, digest_pos;
- for (i = 8; i < 12; i++)
- digest_pos += buf[i];
- digest_pos = (digest_pos % 728) + 12;
+ if (encrypted)
+ digest_pos = ff_rtmp_calc_digest_pos(buf, 772, 728, 776);
+ else
+ digest_pos = ff_rtmp_calc_digest_pos(buf, 8, 728, 12);
- ret = rtmp_calc_digest(buf, RTMP_HANDSHAKE_PACKET_SIZE, digest_pos,
- rtmp_player_key, PLAYER_KEY_OPEN_PART_LEN,
- buf + digest_pos);
+ ret = ff_rtmp_calc_digest(buf, RTMP_HANDSHAKE_PACKET_SIZE, digest_pos,
+ rtmp_player_key, PLAYER_KEY_OPEN_PART_LEN,
+ buf + digest_pos);
if (ret < 0)
return ret;
*/
static int rtmp_validate_digest(uint8_t *buf, int off)
{
- int i, digest_pos = 0;
uint8_t digest[32];
- int ret;
+ int ret, digest_pos;
- for (i = 0; i < 4; i++)
- digest_pos += buf[i + off];
- digest_pos = (digest_pos % 728) + off + 4;
+ digest_pos = ff_rtmp_calc_digest_pos(buf, off, 728, off + 4);
- ret = rtmp_calc_digest(buf, RTMP_HANDSHAKE_PACKET_SIZE, digest_pos,
- rtmp_server_key, SERVER_KEY_OPEN_PART_LEN,
- digest);
+ ret = ff_rtmp_calc_digest(buf, RTMP_HANDSHAKE_PACKET_SIZE, digest_pos,
+ rtmp_server_key, SERVER_KEY_OPEN_PART_LEN,
+ digest);
if (ret < 0)
return ret;
uint8_t serverdata[RTMP_HANDSHAKE_PACKET_SIZE+1];
int i;
int server_pos, client_pos;
- uint8_t digest[32];
- int ret;
+ uint8_t digest[32], signature[32];
+ int ret, type = 0;
av_log(s, AV_LOG_DEBUG, "Handshaking...\n");
// generate handshake packet - 1536 bytes of pseudorandom data
for (i = 9; i <= RTMP_HANDSHAKE_PACKET_SIZE; i++)
tosend[i] = av_lfg_get(&rnd) >> 24;
- client_pos = rtmp_handshake_imprint_with_digest(tosend + 1);
+
+ if (rt->encrypted && CONFIG_FFRTMPCRYPT_PROTOCOL) {
+ /* When the client wants to use RTMPE, we have to change the command
+ * byte to 0x06 which means to use encrypted data and we have to set
+ * the flash version to at least 9.0.115.0. */
+ tosend[0] = 6;
+ tosend[5] = 128;
+ tosend[6] = 0;
+ tosend[7] = 3;
+ tosend[8] = 2;
+
+ /* Initialize the Diffie-Hellmann context and generate the public key
+ * to send to the server. */
+ if ((ret = ff_rtmpe_gen_pub_key(rt->stream, tosend + 1)) < 0)
+ return ret;
+ }
+
+ client_pos = rtmp_handshake_imprint_with_digest(tosend + 1, rt->encrypted);
if (client_pos < 0)
return client_pos;
return ret;
}
+ av_log(s, AV_LOG_DEBUG, "Type answer %d\n", serverdata[0]);
av_log(s, AV_LOG_DEBUG, "Server version %d.%d.%d.%d\n",
serverdata[5], serverdata[6], serverdata[7], serverdata[8]);
return server_pos;
if (!server_pos) {
+ type = 1;
server_pos = rtmp_validate_digest(serverdata + 1, 8);
if (server_pos < 0)
return server_pos;
}
}
- ret = rtmp_calc_digest(tosend + 1 + client_pos, 32, 0, rtmp_server_key,
- sizeof(rtmp_server_key), digest);
+ ret = ff_rtmp_calc_digest(tosend + 1 + client_pos, 32, 0,
+ rtmp_server_key, sizeof(rtmp_server_key),
+ digest);
if (ret < 0)
return ret;
- ret = rtmp_calc_digest(clientdata, RTMP_HANDSHAKE_PACKET_SIZE - 32, 0,
- digest, 32, digest);
+ ret = ff_rtmp_calc_digest(clientdata, RTMP_HANDSHAKE_PACKET_SIZE - 32,
+ 0, digest, 32, signature);
if (ret < 0)
return ret;
- if (memcmp(digest, clientdata + RTMP_HANDSHAKE_PACKET_SIZE - 32, 32)) {
+ if (rt->encrypted && CONFIG_FFRTMPCRYPT_PROTOCOL) {
+ /* Compute the shared secret key sent by the server and initialize
+ * the RC4 encryption. */
+ if ((ret = ff_rtmpe_compute_secret_key(rt->stream, serverdata + 1,
+ tosend + 1, type)) < 0)
+ return ret;
+
+ /* Encrypt the signature received by the server. */
+ ff_rtmpe_encrypt_sig(rt->stream, signature, digest, serverdata[0]);
+ }
+
+ if (memcmp(signature, clientdata + RTMP_HANDSHAKE_PACKET_SIZE - 32, 32)) {
av_log(s, AV_LOG_ERROR, "Signature mismatch\n");
return AVERROR(EIO);
}
for (i = 0; i < RTMP_HANDSHAKE_PACKET_SIZE; i++)
tosend[i] = av_lfg_get(&rnd) >> 24;
- ret = rtmp_calc_digest(serverdata + 1 + server_pos, 32, 0,
- rtmp_player_key, sizeof(rtmp_player_key),
- digest);
+ ret = ff_rtmp_calc_digest(serverdata + 1 + server_pos, 32, 0,
+ rtmp_player_key, sizeof(rtmp_player_key),
+ digest);
if (ret < 0)
return ret;
- ret = rtmp_calc_digest(tosend, RTMP_HANDSHAKE_PACKET_SIZE - 32, 0,
- digest, 32,
- tosend + RTMP_HANDSHAKE_PACKET_SIZE - 32);
+ ret = ff_rtmp_calc_digest(tosend, RTMP_HANDSHAKE_PACKET_SIZE - 32, 0,
+ digest, 32,
+ tosend + RTMP_HANDSHAKE_PACKET_SIZE - 32);
if (ret < 0)
return ret;
+ if (rt->encrypted && CONFIG_FFRTMPCRYPT_PROTOCOL) {
+ /* Encrypt the signature to be send to the server. */
+ ff_rtmpe_encrypt_sig(rt->stream, tosend +
+ RTMP_HANDSHAKE_PACKET_SIZE - 32, digest,
+ serverdata[0]);
+ }
+
// write reply back to the server
if ((ret = ffurl_write(rt->stream, tosend,
RTMP_HANDSHAKE_PACKET_SIZE)) < 0)
return ret;
+
+ if (rt->encrypted && CONFIG_FFRTMPCRYPT_PROTOCOL) {
+ /* Set RC4 keys for encryption and update the keystreams. */
+ if ((ret = ff_rtmpe_update_keystream(rt->stream)) < 0)
+ return ret;
+ }
} else {
+ if (rt->encrypted && CONFIG_FFRTMPCRYPT_PROTOCOL) {
+ /* Compute the shared secret key sent by the server and initialize
+ * the RC4 encryption. */
+ if ((ret = ff_rtmpe_compute_secret_key(rt->stream, serverdata + 1,
+ tosend + 1, 1)) < 0)
+ return ret;
+
+ if (serverdata[0] == 9) {
+ /* Encrypt the signature received by the server. */
+ ff_rtmpe_encrypt_sig(rt->stream, signature, digest,
+ serverdata[0]);
+ }
+ }
+
if ((ret = ffurl_write(rt->stream, serverdata + 1,
RTMP_HANDSHAKE_PACKET_SIZE)) < 0)
return ret;
+
+ if (rt->encrypted && CONFIG_FFRTMPCRYPT_PROTOCOL) {
+ /* Set RC4 keys for encryption and update the keystreams. */
+ if ((ret = ff_rtmpe_update_keystream(rt->stream)) < 0)
+ return ret;
+ }
+ }
+
+ return 0;
+}
+
+static int handle_chunk_size(URLContext *s, RTMPPacket *pkt)
+{
+ RTMPContext *rt = s->priv_data;
+ int ret;
+
+ if (pkt->data_size < 4) {
+ av_log(s, AV_LOG_ERROR,
+ "Too short chunk size change packet (%d)\n",
+ pkt->data_size);
+ return AVERROR_INVALIDDATA;
+ }
+
+ if (!rt->is_input) {
+ if ((ret = ff_rtmp_packet_write(rt->stream, pkt, rt->chunk_size,
+ rt->prev_pkt[1])) < 0)
+ return ret;
+ }
+
+ rt->chunk_size = AV_RB32(pkt->data);
+ if (rt->chunk_size <= 0) {
+ av_log(s, AV_LOG_ERROR, "Incorrect chunk size %d\n", rt->chunk_size);
+ return AVERROR_INVALIDDATA;
+ }
+ av_log(s, AV_LOG_DEBUG, "New chunk size = %d\n", rt->chunk_size);
+
+ return 0;
+}
+
+static int handle_ping(URLContext *s, RTMPPacket *pkt)
+{
+ RTMPContext *rt = s->priv_data;
+ int t, ret;
+
+ if (pkt->data_size < 2) {
+ av_log(s, AV_LOG_ERROR, "Too short ping packet (%d)\n",
+ pkt->data_size);
+ return AVERROR_INVALIDDATA;
+ }
+
+ t = AV_RB16(pkt->data);
+ if (t == 6) {
+ if ((ret = gen_pong(s, rt, pkt)) < 0)
+ return ret;
}
return 0;
}
+static int handle_client_bw(URLContext *s, RTMPPacket *pkt)
+{
+ RTMPContext *rt = s->priv_data;
+
+ if (pkt->data_size < 4) {
+ av_log(s, AV_LOG_ERROR,
+ "Client bandwidth report packet is less than 4 bytes long (%d)\n",
+ pkt->data_size);
+ return AVERROR_INVALIDDATA;
+ }
+
+ rt->client_report_size = AV_RB32(pkt->data);
+ if (rt->client_report_size <= 0) {
+ av_log(s, AV_LOG_ERROR, "Incorrect client bandwidth %d\n",
+ rt->client_report_size);
+ return AVERROR_INVALIDDATA;
+
+ }
+ av_log(s, AV_LOG_DEBUG, "Client bandwidth = %d\n", rt->client_report_size);
+ rt->client_report_size >>= 1;
+
+ return 0;
+}
+
+static int handle_server_bw(URLContext *s, RTMPPacket *pkt)
+{
+ RTMPContext *rt = s->priv_data;
+
+ if (pkt->data_size < 4) {
+ av_log(s, AV_LOG_ERROR,
+ "Too short server bandwidth report packet (%d)\n",
+ pkt->data_size);
+ return AVERROR_INVALIDDATA;
+ }
+
+ rt->server_bw = AV_RB32(pkt->data);
+ if (rt->server_bw <= 0) {
+ av_log(s, AV_LOG_ERROR, "Incorrect server bandwidth %d\n",
+ rt->server_bw);
+ return AVERROR_INVALIDDATA;
+ }
+ av_log(s, AV_LOG_DEBUG, "Server bandwidth = %d\n", rt->server_bw);
+
+ return 0;
+}
+
+static int handle_invoke(URLContext *s, RTMPPacket *pkt)
+{
+ RTMPContext *rt = s->priv_data;
+ int i, t;
+ const uint8_t *data_end = pkt->data + pkt->data_size;
+ char *tracked_method = NULL;
+ int ret = 0;
+
+ //TODO: check for the messages sent for wrong state?
+ if (!memcmp(pkt->data, "\002\000\006_error", 9)) {
+ uint8_t tmpstr[256];
+
+ if (!ff_amf_get_field_value(pkt->data + 9, data_end,
+ "description", tmpstr, sizeof(tmpstr)))
+ av_log(s, AV_LOG_ERROR, "Server error: %s\n",tmpstr);
+ return -1;
+ } else if (!memcmp(pkt->data, "\002\000\007_result", 10)) {
+ GetByteContext gbc;
+ double pkt_id;
+
+ bytestream2_init(&gbc, pkt->data + 10, pkt->data_size);
+ if ((ret = ff_amf_read_number(&gbc, &pkt_id)) < 0)
+ return ret;
+
+ for (i = 0; i < rt->nb_tracked_methods; i++) {
+ if (rt->tracked_methods[i].id != pkt_id)
+ continue;
+
+ tracked_method = rt->tracked_methods[i].name;
+ del_tracked_method(rt, i);
+ break;
+ }
+
+ if (!tracked_method) {
+ /* Ignore this reply when the current method is not tracked. */
+ return 0;
+ }
+
+ if (!memcmp(tracked_method, "connect", 7)) {
+ if (!rt->is_input) {
+ if ((ret = gen_release_stream(s, rt)) < 0)
+ goto invoke_fail;
+
+ if ((ret = gen_fcpublish_stream(s, rt)) < 0)
+ goto invoke_fail;
+ } else {
+ if ((ret = gen_server_bw(s, rt)) < 0)
+ goto invoke_fail;
+ }
+
+ if ((ret = gen_create_stream(s, rt)) < 0)
+ goto invoke_fail;
+
+ if (rt->is_input) {
+ /* Send the FCSubscribe command when the name of live
+ * stream is defined by the user or if it's a live stream. */
+ if (rt->subscribe) {
+ if ((ret = gen_fcsubscribe_stream(s, rt,
+ rt->subscribe)) < 0)
+ goto invoke_fail;
+ } else if (rt->live == -1) {
+ if ((ret = gen_fcsubscribe_stream(s, rt,
+ rt->playpath)) < 0)
+ goto invoke_fail;
+ }
+ }
+ } else if (!memcmp(tracked_method, "createStream", 12)) {
+ //extract a number from the result
+ if (pkt->data[10] || pkt->data[19] != 5 || pkt->data[20]) {
+ av_log(s, AV_LOG_WARNING, "Unexpected reply on connect()\n");
+ } else {
+ rt->main_channel_id = av_int2double(AV_RB64(pkt->data + 21));
+ }
+
+ if (!rt->is_input) {
+ if ((ret = gen_publish(s, rt)) < 0)
+ goto invoke_fail;
+ } else {
+ if ((ret = gen_play(s, rt)) < 0)
+ goto invoke_fail;
+ if ((ret = gen_buffer_time(s, rt)) < 0)
+ goto invoke_fail;
+ }
+ }
+ } else if (!memcmp(pkt->data, "\002\000\010onStatus", 11)) {
+ const uint8_t* ptr = pkt->data + 11;
+ uint8_t tmpstr[256];
+
+ for (i = 0; i < 2; i++) {
+ t = ff_amf_tag_size(ptr, data_end);
+ if (t < 0)
+ return 1;
+ ptr += t;
+ }
+ t = ff_amf_get_field_value(ptr, data_end,
+ "level", tmpstr, sizeof(tmpstr));
+ if (!t && !strcmp(tmpstr, "error")) {
+ if (!ff_amf_get_field_value(ptr, data_end,
+ "description", tmpstr, sizeof(tmpstr)))
+ av_log(s, AV_LOG_ERROR, "Server error: %s\n",tmpstr);
+ return -1;
+ }
+ t = ff_amf_get_field_value(ptr, data_end,
+ "code", tmpstr, sizeof(tmpstr));
+ if (!t && !strcmp(tmpstr, "NetStream.Play.Start")) rt->state = STATE_PLAYING;
+ if (!t && !strcmp(tmpstr, "NetStream.Play.Stop")) rt->state = STATE_STOPPED;
+ if (!t && !strcmp(tmpstr, "NetStream.Play.UnpublishNotify")) rt->state = STATE_STOPPED;
+ if (!t && !strcmp(tmpstr, "NetStream.Publish.Start")) rt->state = STATE_PUBLISHING;
+ } else if (!memcmp(pkt->data, "\002\000\010onBWDone", 11)) {
+ if ((ret = gen_check_bw(s, rt)) < 0)
+ return ret;
+ }
+
+invoke_fail:
+ av_free(tracked_method);
+ return ret;
+}
+
/**
* Parse received packet and possibly perform some action depending on
* the packet contents.
*/
static int rtmp_parse_result(URLContext *s, RTMPContext *rt, RTMPPacket *pkt)
{
- int i, t;
- const uint8_t *data_end = pkt->data + pkt->data_size;
int ret;
#ifdef DEBUG
switch (pkt->type) {
case RTMP_PT_CHUNK_SIZE:
- if (pkt->data_size != 4) {
- av_log(s, AV_LOG_ERROR,
- "Chunk size change packet is not 4 bytes long (%d)\n", pkt->data_size);
- return -1;
- }
- if (!rt->is_input)
- if ((ret = ff_rtmp_packet_write(rt->stream, pkt, rt->chunk_size,
- rt->prev_pkt[1])) < 0)
- return ret;
- rt->chunk_size = AV_RB32(pkt->data);
- if (rt->chunk_size <= 0) {
- av_log(s, AV_LOG_ERROR, "Incorrect chunk size %d\n", rt->chunk_size);
- return -1;
- }
- av_log(s, AV_LOG_DEBUG, "New chunk size = %d\n", rt->chunk_size);
+ if ((ret = handle_chunk_size(s, pkt)) < 0)
+ return ret;
break;
case RTMP_PT_PING:
- t = AV_RB16(pkt->data);
- if (t == 6)
- if ((ret = gen_pong(s, rt, pkt)) < 0)
- return ret;
+ if ((ret = handle_ping(s, pkt)) < 0)
+ return ret;
break;
case RTMP_PT_CLIENT_BW:
- if (pkt->data_size < 4) {
- av_log(s, AV_LOG_ERROR,
- "Client bandwidth report packet is less than 4 bytes long (%d)\n",
- pkt->data_size);
- return -1;
- }
- av_log(s, AV_LOG_DEBUG, "Client bandwidth = %d\n", AV_RB32(pkt->data));
- rt->client_report_size = AV_RB32(pkt->data) >> 1;
+ if ((ret = handle_client_bw(s, pkt)) < 0)
+ return ret;
break;
case RTMP_PT_SERVER_BW:
- rt->server_bw = AV_RB32(pkt->data);
- if (rt->server_bw <= 0) {
- av_log(s, AV_LOG_ERROR, "Incorrect server bandwidth %d\n", rt->server_bw);
- return AVERROR(EINVAL);
- }
- av_log(s, AV_LOG_DEBUG, "Server bandwidth = %d\n", rt->server_bw);
+ if ((ret = handle_server_bw(s, pkt)) < 0)
+ return ret;
break;
case RTMP_PT_INVOKE:
- //TODO: check for the messages sent for wrong state?
- if (!memcmp(pkt->data, "\002\000\006_error", 9)) {
- uint8_t tmpstr[256];
-
- if (!ff_amf_get_field_value(pkt->data + 9, data_end,
- "description", tmpstr, sizeof(tmpstr)))
- av_log(s, AV_LOG_ERROR, "Server error: %s\n",tmpstr);
- return -1;
- } else if (!memcmp(pkt->data, "\002\000\007_result", 10)) {
- switch (rt->state) {
- case STATE_HANDSHAKED:
- if (!rt->is_input) {
- if ((ret = gen_release_stream(s, rt)) < 0)
- return ret;
- if ((ret = gen_fcpublish_stream(s, rt)) < 0)
- return ret;
- rt->state = STATE_RELEASING;
- } else {
- if ((ret = gen_server_bw(s, rt)) < 0)
- return ret;
- rt->state = STATE_CONNECTING;
- }
- if ((ret = gen_create_stream(s, rt)) < 0)
- return ret;
- break;
- case STATE_FCPUBLISH:
- rt->state = STATE_CONNECTING;
- break;
- case STATE_RELEASING:
- rt->state = STATE_FCPUBLISH;
- /* hack for Wowza Media Server, it does not send result for
- * releaseStream and FCPublish calls */
- if (!pkt->data[10]) {
- int pkt_id = av_int2double(AV_RB64(pkt->data + 11));
- if (pkt_id == rt->create_stream_invoke)
- rt->state = STATE_CONNECTING;
- }
- if (rt->state != STATE_CONNECTING)
- break;
- case STATE_CONNECTING:
- //extract a number from the result
- if (pkt->data[10] || pkt->data[19] != 5 || pkt->data[20]) {
- av_log(s, AV_LOG_WARNING, "Unexpected reply on connect()\n");
- } else {
- rt->main_channel_id = av_int2double(AV_RB64(pkt->data + 21));
- }
- if (rt->is_input) {
- if ((ret = gen_play(s, rt)) < 0)
- return ret;
- if ((ret = gen_buffer_time(s, rt)) < 0)
- return ret;
- } else {
- if ((ret = gen_publish(s, rt)) < 0)
- return ret;
- }
- rt->state = STATE_READY;
- break;
- }
- } else if (!memcmp(pkt->data, "\002\000\010onStatus", 11)) {
- const uint8_t* ptr = pkt->data + 11;
- uint8_t tmpstr[256];
-
- for (i = 0; i < 2; i++) {
- t = ff_amf_tag_size(ptr, data_end);
- if (t < 0)
- return 1;
- ptr += t;
- }
- t = ff_amf_get_field_value(ptr, data_end,
- "level", tmpstr, sizeof(tmpstr));
- if (!t && !strcmp(tmpstr, "error")) {
- if (!ff_amf_get_field_value(ptr, data_end,
- "description", tmpstr, sizeof(tmpstr)))
- av_log(s, AV_LOG_ERROR, "Server error: %s\n",tmpstr);
- return -1;
- }
- t = ff_amf_get_field_value(ptr, data_end,
- "code", tmpstr, sizeof(tmpstr));
- if (!t && !strcmp(tmpstr, "NetStream.Play.Start")) rt->state = STATE_PLAYING;
- if (!t && !strcmp(tmpstr, "NetStream.Play.Stop")) rt->state = STATE_STOPPED;
- if (!t && !strcmp(tmpstr, "NetStream.Play.UnpublishNotify")) rt->state = STATE_STOPPED;
- if (!t && !strcmp(tmpstr, "NetStream.Publish.Start")) rt->state = STATE_PUBLISHING;
- } else if (!memcmp(pkt->data, "\002\000\010onBWDone", 11)) {
- if ((ret = gen_check_bw(s, rt)) < 0)
- return ret;
- }
+ if ((ret = handle_invoke(s, pkt)) < 0)
+ return ret;
+ break;
+ case RTMP_PT_VIDEO:
+ case RTMP_PT_AUDIO:
+ case RTMP_PT_METADATA:
+ /* Audio, Video and Metadata packets are parsed in get_packet() */
break;
default:
av_log(s, AV_LOG_VERBOSE, "Unknown packet type received 0x%02X\n", pkt->type);
if (rt->state > STATE_HANDSHAKED)
ret = gen_delete_stream(h, rt);
+ free_tracked_methods(rt);
av_freep(&rt->flv_data);
ffurl_close(rt->stream);
return ret;
char *old_app;
uint8_t buf[2048];
int port;
+ AVDictionary *opts = NULL;
int ret;
rt->is_input = !(flags & AVIO_FLAG_WRITE);
av_url_split(proto, sizeof(proto), NULL, 0, hostname, sizeof(hostname), &port,
path, sizeof(path), s->filename);
- if (port < 0)
- port = RTMP_DEFAULT_PORT;
- ff_url_join(buf, sizeof(buf), "tcp", NULL, hostname, port, NULL);
+ if (!strcmp(proto, "rtmpt") || !strcmp(proto, "rtmpts")) {
+ if (!strcmp(proto, "rtmpts"))
+ av_dict_set(&opts, "ffrtmphttp_tls", "1", 1);
+
+ /* open the http tunneling connection */
+ ff_url_join(buf, sizeof(buf), "ffrtmphttp", NULL, hostname, port, NULL);
+ } else if (!strcmp(proto, "rtmps")) {
+ /* open the tls connection */
+ if (port < 0)
+ port = RTMPS_DEFAULT_PORT;
+ ff_url_join(buf, sizeof(buf), "tls", NULL, hostname, port, NULL);
+ } else if (!strcmp(proto, "rtmpe") || (!strcmp(proto, "rtmpte"))) {
+ if (!strcmp(proto, "rtmpte"))
+ av_dict_set(&opts, "ffrtmpcrypt_tunneling", "1", 1);
+
+ /* open the encrypted connection */
+ ff_url_join(buf, sizeof(buf), "ffrtmpcrypt", NULL, hostname, port, NULL);
+ rt->encrypted = 1;
+ } else {
+ /* open the tcp connection */
+ if (port < 0)
+ port = RTMP_DEFAULT_PORT;
+ ff_url_join(buf, sizeof(buf), "tcp", NULL, hostname, port, NULL);
+ }
if ((ret = ffurl_open(&rt->stream, buf, AVIO_FLAG_READ_WRITE,
- &s->interrupt_callback, NULL)) < 0) {
+ &s->interrupt_callback, &opts)) < 0) {
av_log(s , AV_LOG_ERROR, "Cannot open connection %s\n", buf);
goto fail;
}
return 0;
fail:
+ av_dict_free(&opts);
rtmp_close(s);
return ret;
}
int pktsize, pkttype;
uint32_t ts;
const uint8_t *buf_temp = buf;
+ uint8_t c;
int ret;
do {
if (rt->flv_off == rt->flv_size) {
rt->skip_bytes = 4;
- if ((ret = ff_rtmp_packet_write(rt->stream, &rt->out_pkt,
- rt->chunk_size, rt->prev_pkt[1])) < 0)
+ if ((ret = rtmp_send_packet(rt, &rt->out_pkt, 0)) < 0)
return ret;
- ff_rtmp_packet_destroy(&rt->out_pkt);
rt->flv_size = 0;
rt->flv_off = 0;
rt->flv_header_bytes = 0;
+ rt->flv_nb_packets++;
}
} while (buf_temp - buf < size);
+
+ if (rt->flv_nb_packets < rt->flush_interval)
+ return size;
+ rt->flv_nb_packets = 0;
+
+ /* set stream into nonblocking mode */
+ rt->stream->flags |= AVIO_FLAG_NONBLOCK;
+
+ /* try to read one byte from the stream */
+ ret = ffurl_read(rt->stream, &c, 1);
+
+ /* switch the stream back into blocking mode */
+ rt->stream->flags &= ~AVIO_FLAG_NONBLOCK;
+
+ if (ret == AVERROR(EAGAIN)) {
+ /* no incoming data to handle */
+ return size;
+ } else if (ret < 0) {
+ return ret;
+ } else if (ret == 1) {
+ RTMPPacket rpkt = { 0 };
+
+ if ((ret = ff_rtmp_packet_read_internal(rt->stream, &rpkt,
+ rt->chunk_size,
+ rt->prev_pkt[0], c)) <= 0)
+ return ret;
+
+ if ((ret = rtmp_parse_result(s, rt, &rpkt)) < 0)
+ return ret;
+
+ ff_rtmp_packet_destroy(&rpkt);
+ }
+
return size;
}
{"rtmp_buffer", "Set buffer time in milliseconds. The default is 3000.", OFFSET(client_buffer_time), AV_OPT_TYPE_INT, {3000}, 0, INT_MAX, DEC|ENC},
{"rtmp_conn", "Append arbitrary AMF data to the Connect message", OFFSET(conn), AV_OPT_TYPE_STRING, {.str = NULL }, 0, 0, DEC|ENC},
{"rtmp_flashver", "Version of the Flash plugin used to run the SWF player.", OFFSET(flashver), AV_OPT_TYPE_STRING, {.str = NULL }, 0, 0, DEC|ENC},
+ {"rtmp_flush_interval", "Number of packets flushed in the same request (RTMPT only).", OFFSET(flush_interval), AV_OPT_TYPE_INT, {10}, 0, INT_MAX, ENC},
{"rtmp_live", "Specify that the media is a live stream.", OFFSET(live), AV_OPT_TYPE_INT, {-2}, INT_MIN, INT_MAX, DEC, "rtmp_live"},
{"any", "both", 0, AV_OPT_TYPE_CONST, {-2}, 0, 0, DEC, "rtmp_live"},
{"live", "live stream", 0, AV_OPT_TYPE_CONST, {-1}, 0, 0, DEC, "rtmp_live"},
{"recorded", "recorded stream", 0, AV_OPT_TYPE_CONST, {0}, 0, 0, DEC, "rtmp_live"},
+ {"rtmp_pageurl", "URL of the web page in which the media was embedded. By default no value will be sent.", OFFSET(pageurl), AV_OPT_TYPE_STRING, {.str = NULL }, 0, 0, DEC},
{"rtmp_playpath", "Stream identifier to play or to publish", OFFSET(playpath), AV_OPT_TYPE_STRING, {.str = NULL }, 0, 0, DEC|ENC},
+ {"rtmp_subscribe", "Name of live stream to subscribe to. Defaults to rtmp_playpath.", OFFSET(subscribe), AV_OPT_TYPE_STRING, {.str = NULL }, 0, 0, DEC},
{"rtmp_swfurl", "URL of the SWF player. By default no value will be sent", OFFSET(swfurl), AV_OPT_TYPE_STRING, {.str = NULL }, 0, 0, DEC|ENC},
- {"rtmp_tcurl", "URL of the target stream. Defaults to rtmp://host[:port]/app.", OFFSET(tcurl), AV_OPT_TYPE_STRING, {.str = NULL }, 0, 0, DEC|ENC},
+ {"rtmp_tcurl", "URL of the target stream. Defaults to proto://host[:port]/app.", OFFSET(tcurl), AV_OPT_TYPE_STRING, {.str = NULL }, 0, 0, DEC|ENC},
{ NULL },
};
-static const AVClass rtmp_class = {
- .class_name = "rtmp",
- .item_name = av_default_item_name,
- .option = rtmp_options,
- .version = LIBAVUTIL_VERSION_INT,
+#define RTMP_PROTOCOL(flavor) \
+static const AVClass flavor##_class = { \
+ .class_name = #flavor, \
+ .item_name = av_default_item_name, \
+ .option = rtmp_options, \
+ .version = LIBAVUTIL_VERSION_INT, \
+}; \
+ \
+URLProtocol ff_##flavor##_protocol = { \
+ .name = #flavor, \
+ .url_open = rtmp_open, \
+ .url_read = rtmp_read, \
+ .url_write = rtmp_write, \
+ .url_close = rtmp_close, \
+ .priv_data_size = sizeof(RTMPContext), \
+ .flags = URL_PROTOCOL_FLAG_NETWORK, \
+ .priv_data_class= &flavor##_class, \
};
-URLProtocol ff_rtmp_protocol = {
- .name = "rtmp",
- .url_open = rtmp_open,
- .url_read = rtmp_read,
- .url_write = rtmp_write,
- .url_close = rtmp_close,
- .priv_data_size = sizeof(RTMPContext),
- .flags = URL_PROTOCOL_FLAG_NETWORK,
- .priv_data_class= &rtmp_class,
-};
+
+RTMP_PROTOCOL(rtmp)
+RTMP_PROTOCOL(rtmpe)
+RTMP_PROTOCOL(rtmps)
+RTMP_PROTOCOL(rtmpt)
+RTMP_PROTOCOL(rtmpte)
+RTMP_PROTOCOL(rtmpts)
projects / ffmpeg / blobdiff