Statistics
| Branch: | Revision:

janus-gateway / plugins / janus_echotest.c @ 78955474

History | View | Annotate | Download (27.8 KB)

1
/*! \file   janus_echotest.c
2
 * \author Lorenzo Miniero <lorenzo@meetecho.com>
3
 * \copyright GNU General Public License v3
4
 * \brief  Janus EchoTest plugin
5
 * \details  This is a trivial EchoTest plugin for Janus, just used to
6
 * showcase the plugin interface. A peer attaching to this plugin will
7
 * receive back the same RTP packets and RTCP messages he sends: the
8
 * RTCP messages, of course, would be modified on the way by the gateway
9
 * to make sure they are coherent with the involved SSRCs. In order to
10
 * demonstrate how peer-provided messages can change the behaviour of a
11
 * plugin, this plugin implements a simple API based on three messages:
12
 * 
13
 * 1. a message to enable/disable audio (that is, to tell the plugin
14
 * whether incoming audio RTP packets need to be sent back or discarded);
15
 * 2. a message to enable/disable video (that is, to tell the plugin
16
 * whether incoming video RTP packets need to be sent back or discarded);
17
 * 3. a message to cap the bitrate (which would modify incoming RTCP
18
 * REMB messages before sending them back, in order to trick the peer into
19
 * thinking the available bandwidth is different).
20
 * 
21
 * \section echoapi Echo Test API
22
 * 
23
 * There's a single unnamed request you can send and it's asynchronous,
24
 * which means all responses (successes and errors) will be delivered
25
 * as events with the same transaction. 
26
 * 
27
 * The request has to be formatted as follows:
28
 *
29
\verbatim
30
{
31
        "audio" : true|false,
32
        "video" : true|false,
33
        "bitrate" : <numeric bitrate value>
34
}
35
\endverbatim
36
 *
37
 * \c audio instructs the plugin to do or do not bounce back audio
38
 * frames; \c video does the same for video; \c bitrate caps the
39
 * bandwidth to force on the browser encoding side (e.g., 128000 for
40
 * 128kbps).
41
 * 
42
 * The first request must be sent together with a JSEP offer to
43
 * negotiate a PeerConnection: a JSEP answer will be provided with
44
 * the asynchronous response notification. Subsequent requests (e.g., to
45
 * dynamically manipulate the bitrate while testing) have to be sent
46
 * without any JSEP payload attached.
47
 * 
48
 * A successful request will result in an \c ok event:
49
 * 
50
\verbatim
51
{
52
        "echotest" : "event",
53
        "result": "ok"
54
}
55
\endverbatim
56
 * 
57
 * An error instead will provide both an error code and a more verbose
58
 * description of the cause of the issue:
59
 * 
60
\verbatim
61
{
62
        "echotest" : "event",
63
        "error_code" : <numeric ID, check Macros below>,
64
        "error" : "<error description as a string>"
65
}
66
\endverbatim
67
 *
68
 * If the plugin detects a loss of the associated PeerConnection, a
69
 * "done" notification is triggered to inform the application the Echo
70
 * Test session is over:
71
 * 
72
\verbatim
73
{
74
        "echotest" : "event",
75
        "result": "done"
76
}
77
\endverbatim
78
 *
79
 * \ingroup plugins
80
 * \ref plugins
81
 */
82

    
83
#include "plugin.h"
84

    
85
#include <jansson.h>
86

    
87
#include "../debug.h"
88
#include "../apierror.h"
89
#include "../config.h"
90
#include "../mutex.h"
91
#include "../rtcp.h"
92
#include "../utils.h"
93

    
94

    
95
/* Plugin information */
96
#define JANUS_ECHOTEST_VERSION                        6
97
#define JANUS_ECHOTEST_VERSION_STRING        "0.0.6"
98
#define JANUS_ECHOTEST_DESCRIPTION                "This is a trivial EchoTest plugin for Janus, just used to showcase the plugin interface."
99
#define JANUS_ECHOTEST_NAME                                "JANUS EchoTest plugin"
100
#define JANUS_ECHOTEST_AUTHOR                        "Meetecho s.r.l."
101
#define JANUS_ECHOTEST_PACKAGE                        "janus.plugin.echotest"
102

    
103
/* Plugin methods */
104
janus_plugin *create(void);
105
int janus_echotest_init(janus_callbacks *callback, const char *config_path);
106
void janus_echotest_destroy(void);
107
int janus_echotest_get_api_compatibility(void);
108
int janus_echotest_get_version(void);
109
const char *janus_echotest_get_version_string(void);
110
const char *janus_echotest_get_description(void);
111
const char *janus_echotest_get_name(void);
112
const char *janus_echotest_get_author(void);
113
const char *janus_echotest_get_package(void);
114
void janus_echotest_create_session(janus_plugin_session *handle, int *error);
115
struct janus_plugin_result *janus_echotest_handle_message(janus_plugin_session *handle, char *transaction, char *message, char *sdp_type, char *sdp);
116
void janus_echotest_setup_media(janus_plugin_session *handle);
117
void janus_echotest_incoming_rtp(janus_plugin_session *handle, int video, char *buf, int len);
118
void janus_echotest_incoming_rtcp(janus_plugin_session *handle, int video, char *buf, int len);
119
void janus_echotest_incoming_data(janus_plugin_session *handle, char *buf, int len);
120
void janus_echotest_slow_link(janus_plugin_session *handle, int uplink, int video);
121
void janus_echotest_hangup_media(janus_plugin_session *handle);
122
void janus_echotest_destroy_session(janus_plugin_session *handle, int *error);
123
char *janus_echotest_query_session(janus_plugin_session *handle);
124

    
125
/* Plugin setup */
126
static janus_plugin janus_echotest_plugin =
127
        JANUS_PLUGIN_INIT (
128
                .init = janus_echotest_init,
129
                .destroy = janus_echotest_destroy,
130

    
131
                .get_api_compatibility = janus_echotest_get_api_compatibility,
132
                .get_version = janus_echotest_get_version,
133
                .get_version_string = janus_echotest_get_version_string,
134
                .get_description = janus_echotest_get_description,
135
                .get_name = janus_echotest_get_name,
136
                .get_author = janus_echotest_get_author,
137
                .get_package = janus_echotest_get_package,
138
                
139
                .create_session = janus_echotest_create_session,
140
                .handle_message = janus_echotest_handle_message,
141
                .setup_media = janus_echotest_setup_media,
142
                .incoming_rtp = janus_echotest_incoming_rtp,
143
                .incoming_rtcp = janus_echotest_incoming_rtcp,
144
                .incoming_data = janus_echotest_incoming_data,
145
                .slow_link = janus_echotest_slow_link,
146
                .hangup_media = janus_echotest_hangup_media,
147
                .destroy_session = janus_echotest_destroy_session,
148
                .query_session = janus_echotest_query_session,
149
        );
150

    
151
/* Plugin creator */
152
janus_plugin *create(void) {
153
        JANUS_LOG(LOG_VERB, "%s created!\n", JANUS_ECHOTEST_NAME);
154
        return &janus_echotest_plugin;
155
}
156

    
157

    
158
/* Useful stuff */
159
static gint initialized = 0, stopping = 0;
160
static janus_callbacks *gateway = NULL;
161
static GThread *handler_thread;
162
static GThread *watchdog;
163
static void *janus_echotest_handler(void *data);
164

    
165
typedef struct janus_echotest_message {
166
        janus_plugin_session *handle;
167
        char *transaction;
168
        char *message;
169
        char *sdp_type;
170
        char *sdp;
171
} janus_echotest_message;
172
static GAsyncQueue *messages = NULL;
173

    
174
typedef struct janus_echotest_session {
175
        janus_plugin_session *handle;
176
        gboolean audio_active;
177
        gboolean video_active;
178
        uint64_t bitrate;
179
        guint16 slowlink_count;
180
        gboolean hangingup;
181
        gint64 destroyed;        /* Time at which this session was marked as destroyed */
182
} janus_echotest_session;
183
static GHashTable *sessions;
184
static GList *old_sessions;
185
static janus_mutex sessions_mutex;
186

    
187
void janus_echotest_message_free(janus_echotest_message *msg);
188
void janus_echotest_message_free(janus_echotest_message *msg) {
189
        if(!msg)
190
                return;
191

    
192
        msg->handle = NULL;
193

    
194
        g_free(msg->transaction);
195
        msg->transaction = NULL;
196
        g_free(msg->message);
197
        msg->message = NULL;
198
        g_free(msg->sdp_type);
199
        msg->sdp_type = NULL;
200
        g_free(msg->sdp);
201
        msg->sdp = NULL;
202

    
203
        g_free(msg);
204
}
205

    
206

    
207
/* Error codes */
208
#define JANUS_ECHOTEST_ERROR_NO_MESSAGE                        411
209
#define JANUS_ECHOTEST_ERROR_INVALID_JSON                412
210
#define JANUS_ECHOTEST_ERROR_INVALID_ELEMENT        413
211

    
212

    
213
/* EchoTest watchdog/garbage collector (sort of) */
214
void *janus_echotest_watchdog(void *data);
215
void *janus_echotest_watchdog(void *data) {
216
        JANUS_LOG(LOG_INFO, "EchoTest watchdog started\n");
217
        gint64 now = 0;
218
        while(g_atomic_int_get(&initialized) && !g_atomic_int_get(&stopping)) {
219
                janus_mutex_lock(&sessions_mutex);
220
                /* Iterate on all the sessions */
221
                now = janus_get_monotonic_time();
222
                if(old_sessions != NULL) {
223
                        GList *sl = old_sessions;
224
                        JANUS_LOG(LOG_HUGE, "Checking %d old EchoTest sessions...\n", g_list_length(old_sessions));
225
                        while(sl) {
226
                                janus_echotest_session *session = (janus_echotest_session *)sl->data;
227
                                if(!session) {
228
                                        sl = sl->next;
229
                                        continue;
230
                                }
231
                                if(now-session->destroyed >= 5*G_USEC_PER_SEC) {
232
                                        /* We're lazy and actually get rid of the stuff only after a few seconds */
233
                                        JANUS_LOG(LOG_VERB, "Freeing old EchoTest session\n");
234
                                        GList *rm = sl->next;
235
                                        old_sessions = g_list_delete_link(old_sessions, sl);
236
                                        sl = rm;
237
                                        session->handle = NULL;
238
                                        g_free(session);
239
                                        session = NULL;
240
                                        continue;
241
                                }
242
                                sl = sl->next;
243
                        }
244
                }
245
                janus_mutex_unlock(&sessions_mutex);
246
                g_usleep(500000);
247
        }
248
        JANUS_LOG(LOG_INFO, "EchoTest watchdog stopped\n");
249
        return NULL;
250
}
251

    
252

    
253
/* Plugin implementation */
254
int janus_echotest_init(janus_callbacks *callback, const char *config_path) {
255
        if(g_atomic_int_get(&stopping)) {
256
                /* Still stopping from before */
257
                return -1;
258
        }
259
        if(callback == NULL || config_path == NULL) {
260
                /* Invalid arguments */
261
                return -1;
262
        }
263

    
264
        /* Read configuration */
265
        char filename[255];
266
        g_snprintf(filename, 255, "%s/%s.cfg", config_path, JANUS_ECHOTEST_PACKAGE);
267
        JANUS_LOG(LOG_VERB, "Configuration file: %s\n", filename);
268
        janus_config *config = janus_config_parse(filename);
269
        if(config != NULL)
270
                janus_config_print(config);
271
        /* This plugin actually has nothing to configure... */
272
        janus_config_destroy(config);
273
        config = NULL;
274
        
275
        sessions = g_hash_table_new(NULL, NULL);
276
        janus_mutex_init(&sessions_mutex);
277
        messages = g_async_queue_new_full((GDestroyNotify) janus_echotest_message_free);
278
        /* This is the callback we'll need to invoke to contact the gateway */
279
        gateway = callback;
280
        g_atomic_int_set(&initialized, 1);
281

    
282
        GError *error = NULL;
283
        /* Start the sessions watchdog */
284
        watchdog = g_thread_try_new("etest watchdog", &janus_echotest_watchdog, NULL, &error);
285
        if(error != NULL) {
286
                g_atomic_int_set(&initialized, 0);
287
                JANUS_LOG(LOG_ERR, "Got error %d (%s) trying to launch the EchoTest watchdog thread...\n", error->code, error->message ? error->message : "??");
288
                return -1;
289
        }
290
        /* Launch the thread that will handle incoming messages */
291
        handler_thread = g_thread_try_new("janus echotest handler", janus_echotest_handler, NULL, &error);
292
        if(error != NULL) {
293
                g_atomic_int_set(&initialized, 0);
294
                JANUS_LOG(LOG_ERR, "Got error %d (%s) trying to launch the EchoTest handler thread...\n", error->code, error->message ? error->message : "??");
295
                return -1;
296
        }
297
        JANUS_LOG(LOG_INFO, "%s initialized!\n", JANUS_ECHOTEST_NAME);
298
        return 0;
299
}
300

    
301
void janus_echotest_destroy(void) {
302
        if(!g_atomic_int_get(&initialized))
303
                return;
304
        g_atomic_int_set(&stopping, 1);
305

    
306
        if(handler_thread != NULL) {
307
                g_thread_join(handler_thread);
308
                handler_thread = NULL;
309
        }
310
        if(watchdog != NULL) {
311
                g_thread_join(watchdog);
312
                watchdog = NULL;
313
        }
314

    
315
        /* FIXME We should destroy the sessions cleanly */
316
        janus_mutex_lock(&sessions_mutex);
317
        g_hash_table_destroy(sessions);
318
        janus_mutex_unlock(&sessions_mutex);
319
        g_async_queue_unref(messages);
320
        messages = NULL;
321
        sessions = NULL;
322

    
323
        g_atomic_int_set(&initialized, 0);
324
        g_atomic_int_set(&stopping, 0);
325
        JANUS_LOG(LOG_INFO, "%s destroyed!\n", JANUS_ECHOTEST_NAME);
326
}
327

    
328
int janus_echotest_get_api_compatibility(void) {
329
        /* Important! This is what your plugin MUST always return: don't lie here or bad things will happen */
330
        return JANUS_PLUGIN_API_VERSION;
331
}
332

    
333
int janus_echotest_get_version(void) {
334
        return JANUS_ECHOTEST_VERSION;
335
}
336

    
337
const char *janus_echotest_get_version_string(void) {
338
        return JANUS_ECHOTEST_VERSION_STRING;
339
}
340

    
341
const char *janus_echotest_get_description(void) {
342
        return JANUS_ECHOTEST_DESCRIPTION;
343
}
344

    
345
const char *janus_echotest_get_name(void) {
346
        return JANUS_ECHOTEST_NAME;
347
}
348

    
349
const char *janus_echotest_get_author(void) {
350
        return JANUS_ECHOTEST_AUTHOR;
351
}
352

    
353
const char *janus_echotest_get_package(void) {
354
        return JANUS_ECHOTEST_PACKAGE;
355
}
356

    
357
void janus_echotest_create_session(janus_plugin_session *handle, int *error) {
358
        if(g_atomic_int_get(&stopping) || !g_atomic_int_get(&initialized)) {
359
                *error = -1;
360
                return;
361
        }        
362
        janus_echotest_session *session = (janus_echotest_session *)calloc(1, sizeof(janus_echotest_session));
363
        if(session == NULL) {
364
                JANUS_LOG(LOG_FATAL, "Memory error!\n");
365
                *error = -2;
366
                return;
367
        }
368
        session->handle = handle;
369
        session->audio_active = TRUE;
370
        session->video_active = TRUE;
371
        session->bitrate = 0;        /* No limit */
372
        session->destroyed = 0;
373
        handle->plugin_handle = session;
374
        janus_mutex_lock(&sessions_mutex);
375
        g_hash_table_insert(sessions, handle, session);
376
        janus_mutex_unlock(&sessions_mutex);
377

    
378
        return;
379
}
380

    
381
void janus_echotest_destroy_session(janus_plugin_session *handle, int *error) {
382
        if(g_atomic_int_get(&stopping) || !g_atomic_int_get(&initialized)) {
383
                *error = -1;
384
                return;
385
        }        
386
        janus_echotest_session *session = (janus_echotest_session *)handle->plugin_handle;
387
        if(!session) {
388
                JANUS_LOG(LOG_ERR, "No session associated with this handle...\n");
389
                *error = -2;
390
                return;
391
        }
392
        JANUS_LOG(LOG_VERB, "Removing Echo Test session...\n");
393
        janus_mutex_lock(&sessions_mutex);
394
        if(!session->destroyed) {
395
                session->destroyed = janus_get_monotonic_time();
396
                g_hash_table_remove(sessions, handle);
397
                /* Cleaning up and removing the session is done in a lazy way */
398
                old_sessions = g_list_append(old_sessions, session);
399
        }
400
        janus_mutex_unlock(&sessions_mutex);
401
        return;
402
}
403

    
404
char *janus_echotest_query_session(janus_plugin_session *handle) {
405
        if(g_atomic_int_get(&stopping) || !g_atomic_int_get(&initialized)) {
406
                return NULL;
407
        }        
408
        janus_echotest_session *session = (janus_echotest_session *)handle->plugin_handle;
409
        if(!session) {
410
                JANUS_LOG(LOG_ERR, "No session associated with this handle...\n");
411
                return NULL;
412
        }
413
        /* In the echo test, every session is the same: we just provide some configure info */
414
        json_t *info = json_object();
415
        json_object_set_new(info, "audio_active", json_string(session->audio_active ? "true" : "false"));
416
        json_object_set_new(info, "video_active", json_string(session->video_active ? "true" : "false"));
417
        json_object_set_new(info, "bitrate", json_integer(session->bitrate));
418
        json_object_set_new(info, "slowlink_count", json_integer(session->slowlink_count));
419
        json_object_set_new(info, "destroyed", json_integer(session->destroyed));
420
        char *info_text = json_dumps(info, JSON_INDENT(3) | JSON_PRESERVE_ORDER);
421
        json_decref(info);
422
        return info_text;
423
}
424

    
425
struct janus_plugin_result *janus_echotest_handle_message(janus_plugin_session *handle, char *transaction, char *message, char *sdp_type, char *sdp) {
426
        if(g_atomic_int_get(&stopping) || !g_atomic_int_get(&initialized))
427
                return janus_plugin_result_new(JANUS_PLUGIN_ERROR, g_atomic_int_get(&stopping) ? "Shutting down" : "Plugin not initialized");
428
        janus_echotest_message *msg = calloc(1, sizeof(janus_echotest_message));
429
        if(msg == NULL) {
430
                JANUS_LOG(LOG_FATAL, "Memory error!\n");
431
                return janus_plugin_result_new(JANUS_PLUGIN_ERROR, "Memory error");
432
        }
433
        msg->handle = handle;
434
        msg->transaction = transaction;
435
        msg->message = message;
436
        msg->sdp_type = sdp_type;
437
        msg->sdp = sdp;
438
        g_async_queue_push(messages, msg);
439

    
440
        /* All the requests to this plugin are handled asynchronously */
441
        return janus_plugin_result_new(JANUS_PLUGIN_OK_WAIT, "I'm taking my time!");
442
}
443

    
444
void janus_echotest_setup_media(janus_plugin_session *handle) {
445
        JANUS_LOG(LOG_INFO, "WebRTC media is now available\n");
446
        if(g_atomic_int_get(&stopping) || !g_atomic_int_get(&initialized))
447
                return;
448
        janus_echotest_session *session = (janus_echotest_session *)handle->plugin_handle;        
449
        if(!session) {
450
                JANUS_LOG(LOG_ERR, "No session associated with this handle...\n");
451
                return;
452
        }
453
        if(session->destroyed)
454
                return;
455
        /* We really don't care, as we only send RTP/RTCP we get in the first place back anyway */
456
}
457

    
458
void janus_echotest_incoming_rtp(janus_plugin_session *handle, int video, char *buf, int len) {
459
        if(handle == NULL || handle->stopped || g_atomic_int_get(&stopping) || !g_atomic_int_get(&initialized))
460
                return;
461
        /* Simple echo test */
462
        if(gateway) {
463
                /* Honour the audio/video active flags */
464
                janus_echotest_session *session = (janus_echotest_session *)handle->plugin_handle;        
465
                if(!session) {
466
                        JANUS_LOG(LOG_ERR, "No session associated with this handle...\n");
467
                        return;
468
                }
469
                if(session->destroyed)
470
                        return;
471
                if((!video && session->audio_active) || (video && session->video_active)) {
472
                        gateway->relay_rtp(handle, video, buf, len);
473
                }
474
        }
475
}
476

    
477
void janus_echotest_incoming_rtcp(janus_plugin_session *handle, int video, char *buf, int len) {
478
        if(handle == NULL || handle->stopped || g_atomic_int_get(&stopping) || !g_atomic_int_get(&initialized))
479
                return;
480
        /* Simple echo test */
481
        if(gateway) {
482
                janus_echotest_session *session = (janus_echotest_session *)handle->plugin_handle;        
483
                if(!session) {
484
                        JANUS_LOG(LOG_ERR, "No session associated with this handle...\n");
485
                        return;
486
                }
487
                if(session->destroyed)
488
                        return;
489
                if(session->bitrate > 0)
490
                        janus_rtcp_cap_remb(buf, len, session->bitrate);
491
                gateway->relay_rtcp(handle, video, buf, len);
492
        }
493
}
494

    
495
void janus_echotest_incoming_data(janus_plugin_session *handle, char *buf, int len) {
496
        if(handle == NULL || handle->stopped || g_atomic_int_get(&stopping) || !g_atomic_int_get(&initialized))
497
                return;
498
        /* Simple echo test */
499
        if(gateway) {
500
                janus_echotest_session *session = (janus_echotest_session *)handle->plugin_handle;        
501
                if(!session) {
502
                        JANUS_LOG(LOG_ERR, "No session associated with this handle...\n");
503
                        return;
504
                }
505
                if(session->destroyed)
506
                        return;
507
                if(buf == NULL || len <= 0)
508
                        return;
509
                char *text = calloc(len+1, sizeof(char));
510
                memcpy(text, buf, len);
511
                *(text+len) = '\0';
512
                JANUS_LOG(LOG_VERB, "Got a DataChannel message (%zu bytes) to bounce back: %s\n", strlen(text), text);
513
                char reply[1<<16];
514
                memset(reply, 0, 1<<16);
515
                g_snprintf(reply, 1<<16, "Janus EchoTest here! You wrote: %s", text);
516
                free(text);
517
                gateway->relay_data(handle, reply, strlen(reply));
518
        }
519
}
520

    
521
void janus_echotest_slow_link(janus_plugin_session *handle, int uplink, int video) {
522
        /* The core is informing us that our peer got or sent too many NACKs, are we pushing media too hard? */
523
        if(handle == NULL || handle->stopped || g_atomic_int_get(&stopping) || !g_atomic_int_get(&initialized))
524
                return;
525
        janus_echotest_session *session = (janus_echotest_session *)handle->plugin_handle;        
526
        if(!session) {
527
                JANUS_LOG(LOG_ERR, "No session associated with this handle...\n");
528
                return;
529
        }
530
        if(session->destroyed)
531
                return;
532
        session->slowlink_count++;
533
        if(uplink && !video && !session->audio_active) {
534
                /* We're not relaying audio and the peer is expecting it, so NACKs are normal */
535
                JANUS_LOG(LOG_VERB, "Getting a lot of NACKs (slow uplink) for audio, but that's expected, a configure disabled the audio forwarding\n");
536
        } else if(uplink && video && !session->video_active) {
537
                /* We're not relaying video and the peer is expecting it, so NACKs are normal */
538
                JANUS_LOG(LOG_VERB, "Getting a lot of NACKs (slow uplink) for video, but that's expected, a configure disabled the video forwarding\n");
539
        } else {
540
                /* Slow uplink or downlink, maybe we set the bitrate cap too high? */
541
                if(video) {
542
                        /* Halve the bitrate, but don't go too low... */
543
                        session->bitrate = session->bitrate > 0 ? session->bitrate : 512*1024;
544
                        session->bitrate = session->bitrate/2;
545
                        if(session->bitrate < 64*1024)
546
                                session->bitrate = 64*1024;
547
                        JANUS_LOG(LOG_WARN, "Getting a lot of NACKs (slow %s) for %s, forcing a lower REMB: %"SCNu64"\n",
548
                                uplink ? "uplink" : "downlink", video ? "video" : "audio", session->bitrate);
549
                        /* ... and send a new REMB back */
550
                        char rtcpbuf[200];
551
                        memset(rtcpbuf, 0, 200);
552
                        /* FIXME First put a RR (fake)... */
553
                        int rrlen = 32;
554
                        rtcp_rr *rr = (rtcp_rr *)&rtcpbuf;
555
                        rr->header.version = 2;
556
                        rr->header.type = RTCP_RR;
557
                        rr->header.rc = 1;
558
                        rr->header.length = htons((rrlen/4)-1);
559
                        /* ... then put a SDES... */
560
                        int sdeslen = janus_rtcp_sdes((char *)(&rtcpbuf)+rrlen, 200-rrlen, "janusvideo", 10);
561
                        if(sdeslen > 0) {
562
                                /* ... and then finally a REMB */
563
                                janus_rtcp_remb((char *)(&rtcpbuf)+rrlen+sdeslen, 24, session->bitrate);
564
                                gateway->relay_rtcp(handle, 1, rtcpbuf, rrlen+sdeslen+24);
565
                        }
566
                        /* As a last thing, notify the user about this */
567
                        json_t *event = json_object();
568
                        json_object_set_new(event, "echotest", json_string("event"));
569
                        json_t *result = json_object();
570
                        json_object_set_new(result, "status", json_string("slow_link"));
571
                        json_object_set_new(result, "bitrate", json_integer(session->bitrate));
572
                        json_object_set_new(event, "result", result);
573
                        char *event_text = json_dumps(event, JSON_INDENT(3) | JSON_PRESERVE_ORDER);
574
                        json_decref(event);
575
                        json_decref(result);
576
                        event = NULL;
577
                        gateway->push_event(session->handle, &janus_echotest_plugin, NULL, event_text, NULL, NULL);
578
                        g_free(event_text);
579
                }
580
        }
581
}
582

    
583
void janus_echotest_hangup_media(janus_plugin_session *handle) {
584
        JANUS_LOG(LOG_INFO, "No WebRTC media anymore\n");
585
        if(g_atomic_int_get(&stopping) || !g_atomic_int_get(&initialized))
586
                return;
587
        janus_echotest_session *session = (janus_echotest_session *)handle->plugin_handle;
588
        if(!session) {
589
                JANUS_LOG(LOG_ERR, "No session associated with this handle...\n");
590
                return;
591
        }
592
        if(session->destroyed || session->hangingup)
593
                return;
594
        session->hangingup = TRUE;
595
        /* Send an event to the browser and tell it's over */
596
        json_t *event = json_object();
597
        json_object_set_new(event, "echotest", json_string("event"));
598
        json_object_set_new(event, "result", json_string("done"));
599
        char *event_text = json_dumps(event, JSON_INDENT(3) | JSON_PRESERVE_ORDER);
600
        json_decref(event);
601
        JANUS_LOG(LOG_VERB, "Pushing event: %s\n", event_text);
602
        int ret = gateway->push_event(handle, &janus_echotest_plugin, NULL, event_text, NULL, NULL);
603
        JANUS_LOG(LOG_VERB, "  >> %d (%s)\n", ret, janus_get_api_error(ret));
604
        g_free(event_text);
605
        /* Reset controls */
606
        session->audio_active = TRUE;
607
        session->video_active = TRUE;
608
        session->bitrate = 0;
609
        /* Done */
610
        session->hangingup = FALSE;
611
}
612

    
613
/* Thread to handle incoming messages */
614
static void *janus_echotest_handler(void *data) {
615
        JANUS_LOG(LOG_VERB, "Joining EchoTest handler thread\n");
616
        janus_echotest_message *msg = NULL;
617
        int error_code = 0;
618
        char *error_cause = calloc(512, sizeof(char));
619
        if(error_cause == NULL) {
620
                JANUS_LOG(LOG_FATAL, "Memory error!\n");
621
                return NULL;
622
        }
623
        json_t *root = NULL;
624
        while(g_atomic_int_get(&initialized) && !g_atomic_int_get(&stopping)) {
625
                if(!messages || (msg = g_async_queue_try_pop(messages)) == NULL) {
626
                        usleep(50000);
627
                        continue;
628
                }
629
                janus_echotest_session *session = (janus_echotest_session *)msg->handle->plugin_handle;
630
                if(!session) {
631
                        JANUS_LOG(LOG_ERR, "No session associated with this handle...\n");
632
                        janus_echotest_message_free(msg);
633
                        continue;
634
                }
635
                if(session->destroyed) {
636
                        janus_echotest_message_free(msg);
637
                        continue;
638
                }
639
                /* Handle request */
640
                error_code = 0;
641
                root = NULL;
642
                JANUS_LOG(LOG_VERB, "Handling message: %s\n", msg->message);
643
                if(msg->message == NULL) {
644
                        JANUS_LOG(LOG_ERR, "No message??\n");
645
                        error_code = JANUS_ECHOTEST_ERROR_NO_MESSAGE;
646
                        g_snprintf(error_cause, 512, "%s", "No message??");
647
                        goto error;
648
                }
649
                json_error_t error;
650
                root = json_loads(msg->message, 0, &error);
651
                if(!root) {
652
                        JANUS_LOG(LOG_ERR, "JSON error: on line %d: %s\n", error.line, error.text);
653
                        error_code = JANUS_ECHOTEST_ERROR_INVALID_JSON;
654
                        g_snprintf(error_cause, 512, "JSON error: on line %d: %s", error.line, error.text);
655
                        goto error;
656
                }
657
                if(!json_is_object(root)) {
658
                        JANUS_LOG(LOG_ERR, "JSON error: not an object\n");
659
                        error_code = JANUS_ECHOTEST_ERROR_INVALID_JSON;
660
                        g_snprintf(error_cause, 512, "JSON error: not an object");
661
                        goto error;
662
                }
663
                json_t *audio = json_object_get(root, "audio");
664
                if(audio && !json_is_boolean(audio)) {
665
                        JANUS_LOG(LOG_ERR, "Invalid element (audio should be a boolean)\n");
666
                        error_code = JANUS_ECHOTEST_ERROR_INVALID_ELEMENT;
667
                        g_snprintf(error_cause, 512, "Invalid value (audio should be a boolean)");
668
                        goto error;
669
                }
670
                json_t *video = json_object_get(root, "video");
671
                if(video && !json_is_boolean(video)) {
672
                        JANUS_LOG(LOG_ERR, "Invalid element (video should be a boolean)\n");
673
                        error_code = JANUS_ECHOTEST_ERROR_INVALID_ELEMENT;
674
                        g_snprintf(error_cause, 512, "Invalid value (video should be a boolean)");
675
                        goto error;
676
                }
677
                json_t *bitrate = json_object_get(root, "bitrate");
678
                if(bitrate && (!json_is_integer(bitrate) || json_integer_value(bitrate) < 0)) {
679
                        JANUS_LOG(LOG_ERR, "Invalid element (bitrate should be a positive integer)\n");
680
                        error_code = JANUS_ECHOTEST_ERROR_INVALID_ELEMENT;
681
                        g_snprintf(error_cause, 512, "Invalid value (bitrate should be a positive integer)");
682
                        goto error;
683
                }
684
                if(audio) {
685
                        session->audio_active = json_is_true(audio);
686
                        JANUS_LOG(LOG_VERB, "Setting audio property: %s\n", session->audio_active ? "true" : "false");
687
                }
688
                if(video) {
689
                        if(!session->video_active && json_is_true(video)) {
690
                                /* Send a PLI */
691
                                JANUS_LOG(LOG_VERB, "Just (re-)enabled video, sending a PLI to recover it\n");
692
                                char buf[12];
693
                                memset(buf, 0, 12);
694
                                janus_rtcp_pli((char *)&buf, 12);
695
                                gateway->relay_rtcp(session->handle, 1, buf, 12);
696
                        }
697
                        session->video_active = json_is_true(video);
698
                        JANUS_LOG(LOG_VERB, "Setting video property: %s\n", session->video_active ? "true" : "false");
699
                }
700
                if(bitrate) {
701
                        session->bitrate = json_integer_value(bitrate);
702
                        JANUS_LOG(LOG_VERB, "Setting video bitrate: %"SCNu64"\n", session->bitrate);
703
                        if(session->bitrate > 0) {
704
                                /* FIXME Generate a new REMB (especially useful for Firefox, which doesn't send any we can cap later) */
705
                                char buf[24];
706
                                memset(buf, 0, 24);
707
                                janus_rtcp_remb((char *)&buf, 24, session->bitrate);
708
                                JANUS_LOG(LOG_VERB, "Sending REMB\n");
709
                                gateway->relay_rtcp(session->handle, 1, buf, 24);
710
                                /* FIXME How should we handle a subsequent "no limit" bitrate? */
711
                        }
712
                }
713
                /* Any SDP to handle? */
714
                if(msg->sdp) {
715
                        JANUS_LOG(LOG_VERB, "This is involving a negotiation (%s) as well:\n%s\n", msg->sdp_type, msg->sdp);
716
                }
717

    
718
                if(!audio && !video && !bitrate && !msg->sdp) {
719
                        JANUS_LOG(LOG_ERR, "No supported attributes (audio, video, bitrate, jsep) found\n");
720
                        error_code = JANUS_ECHOTEST_ERROR_INVALID_ELEMENT;
721
                        g_snprintf(error_cause, 512, "Message error: no supported attributes (audio, video, bitrate, jsep) found");
722
                        goto error;
723
                }
724

    
725
                json_decref(root);
726
                /* Prepare JSON event */
727
                json_t *event = json_object();
728
                json_object_set_new(event, "echotest", json_string("event"));
729
                json_object_set_new(event, "result", json_string("ok"));
730
                char *event_text = json_dumps(event, JSON_INDENT(3) | JSON_PRESERVE_ORDER);
731
                json_decref(event);
732
                JANUS_LOG(LOG_VERB, "Pushing event: %s\n", event_text);
733
                if(!msg->sdp) {
734
                        int ret = gateway->push_event(msg->handle, &janus_echotest_plugin, msg->transaction, event_text, NULL, NULL);
735
                        JANUS_LOG(LOG_VERB, "  >> %d (%s)\n", ret, janus_get_api_error(ret));
736
                } else {
737
                        /* Forward the same offer to the gateway, to start the echo test */
738
                        const char *type = NULL;
739
                        if(!strcasecmp(msg->sdp_type, "offer"))
740
                                type = "answer";
741
                        if(!strcasecmp(msg->sdp_type, "answer"))
742
                                type = "offer";
743
                        /* Any media direction that needs to be fixed? */
744
                        char *sdp = g_strdup(msg->sdp);
745
                        if(strstr(sdp, "a=recvonly")) {
746
                                /* Turn recvonly to inactive, as we simply bounce media back */
747
                                sdp = janus_string_replace(sdp, "a=recvonly", "a=inactive");
748
                        } else if(strstr(sdp, "a=sendonly")) {
749
                                /* Turn sendonly to recvonly */
750
                                sdp = janus_string_replace(sdp, "a=sendonly", "a=recvonly");
751
                                /* FIXME We should also actually not echo this media back, though... */
752
                        }
753
                        /* How long will the gateway take to push the event? */
754
                        gint64 start = janus_get_monotonic_time();
755
                        int res = gateway->push_event(msg->handle, &janus_echotest_plugin, msg->transaction, event_text, type, sdp);
756
                        JANUS_LOG(LOG_VERB, "  >> Pushing event: %d (took %"SCNu64" us)\n",
757
                                res, janus_get_monotonic_time()-start);
758
                        g_free(sdp);
759
                }
760
                g_free(event_text);
761
                janus_echotest_message_free(msg);
762
                continue;
763
                
764
error:
765
                {
766
                        if(root != NULL)
767
                                json_decref(root);
768
                        /* Prepare JSON error event */
769
                        json_t *event = json_object();
770
                        json_object_set_new(event, "echotest", json_string("event"));
771
                        json_object_set_new(event, "error_code", json_integer(error_code));
772
                        json_object_set_new(event, "error", json_string(error_cause));
773
                        char *event_text = json_dumps(event, JSON_INDENT(3) | JSON_PRESERVE_ORDER);
774
                        json_decref(event);
775
                        JANUS_LOG(LOG_VERB, "Pushing event: %s\n", event_text);
776
                        int ret = gateway->push_event(msg->handle, &janus_echotest_plugin, msg->transaction, event_text, NULL, NULL);
777
                        JANUS_LOG(LOG_VERB, "  >> %d (%s)\n", ret, janus_get_api_error(ret));
778
                        g_free(event_text);
779
                        janus_echotest_message_free(msg);
780
                }
781
        }
782
        g_free(error_cause);
783
        JANUS_LOG(LOG_VERB, "Leaving EchoTest handler thread\n");
784
        return NULL;
785
}