gst/rtsp-server/rtsp-session-pool.c

   1 /* GStreamer
   2  * Copyright (C) 2008 Wim Taymans <wim.taymans at gmail.com>
   3  *
   4  * This library is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU Library General Public
   6  * License as published by the Free Software Foundation; either
   7  * version 2 of the License, or (at your option) any later version.
   8  *
   9  * This library is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * Library General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Library General Public
  15  * License along with this library; if not, write to the
  16  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  17  * Boston, MA 02110-1301, USA.
  18  */
  19
  20 #include "rtsp-session-pool.h"
  21
  22 #define DEFAULT_MAX_SESSIONS 0
  23
  24 enum
  25 {
  26   PROP_0,
  27   PROP_MAX_SESSIONS,
  28   PROP_LAST
  29 };
  30
  31 static const gchar session_id_charset[] =
  32     { 'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o',
  33   'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z', 'A', 'B', 'C', 'D',
  34   'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', 'N', 'O', 'P', 'Q', 'R', 'S',
  35   'T', 'U', 'V', 'W', 'X', 'Y', 'Z', '0', '1', '2', '3', '4', '5', '6', '7',
  36   '8', '9', '$', '-', '_', '.', '+'
  37 };
  38
  39 GST_DEBUG_CATEGORY_STATIC (rtsp_session_debug);
  40 #define GST_CAT_DEFAULT rtsp_session_debug
  41
  42 static void gst_rtsp_session_pool_get_property (GObject * object, guint propid,
  43     GValue * value, GParamSpec * pspec);
  44 static void gst_rtsp_session_pool_set_property (GObject * object, guint propid,
  45     const GValue * value, GParamSpec * pspec);
  46 static void gst_rtsp_session_pool_finalize (GObject * object);
  47
  48 static gchar *create_session_id (GstRTSPSessionPool * pool);
  49
  50 G_DEFINE_TYPE (GstRTSPSessionPool, gst_rtsp_session_pool, G_TYPE_OBJECT);
  51
  52 static void
  53 gst_rtsp_session_pool_class_init (GstRTSPSessionPoolClass * klass)
  54 {
  55   GObjectClass *gobject_class;
  56
  57   gobject_class = G_OBJECT_CLASS (klass);
  58
  59   gobject_class->get_property = gst_rtsp_session_pool_get_property;
  60   gobject_class->set_property = gst_rtsp_session_pool_set_property;
  61   gobject_class->finalize = gst_rtsp_session_pool_finalize;
  62
  63   g_object_class_install_property (gobject_class, PROP_MAX_SESSIONS,
  64       g_param_spec_uint ("max-sessions", "Max Sessions",
  65           "the maximum amount of sessions (0 = unlimited)",
  66           0, G_MAXUINT, DEFAULT_MAX_SESSIONS,
  67           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
  68
  69   klass->create_session_id = create_session_id;
  70
  71   GST_DEBUG_CATEGORY_INIT (rtsp_session_debug, "rtspsessionpool", 0,
  72       "GstRTSPSessionPool");
  73 }
  74
  75 static void
  76 gst_rtsp_session_pool_init (GstRTSPSessionPool * pool)
  77 {
  78   g_mutex_init (&pool->lock);
  79   pool->sessions = g_hash_table_new_full (g_str_hash, g_str_equal,
  80       NULL, g_object_unref);
  81   pool->max_sessions = DEFAULT_MAX_SESSIONS;
  82 }
  83
  84 static void
  85 gst_rtsp_session_pool_finalize (GObject * object)
  86 {
  87   GstRTSPSessionPool *pool = GST_RTSP_SESSION_POOL (object);
  88
  89   g_mutex_clear (&pool->lock);
  90   g_hash_table_unref (pool->sessions);
  91
  92   G_OBJECT_CLASS (gst_rtsp_session_pool_parent_class)->finalize (object);
  93 }
  94
  95 static void
  96 gst_rtsp_session_pool_get_property (GObject * object, guint propid,
  97     GValue * value, GParamSpec * pspec)
  98 {
  99   GstRTSPSessionPool *pool = GST_RTSP_SESSION_POOL (object);
 100
 101   switch (propid) {
 102     case PROP_MAX_SESSIONS:
 103       g_value_set_uint (value, gst_rtsp_session_pool_get_max_sessions (pool));
 104       break;
 105     default:
 106       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, propid, pspec);
 107       break;
 108   }
 109 }
 110
 111 static void
 112 gst_rtsp_session_pool_set_property (GObject * object, guint propid,
 113     const GValue * value, GParamSpec * pspec)
 114 {
 115   GstRTSPSessionPool *pool = GST_RTSP_SESSION_POOL (object);
 116
 117   switch (propid) {
 118     case PROP_MAX_SESSIONS:
 119       gst_rtsp_session_pool_set_max_sessions (pool, g_value_get_uint (value));
 120       break;
 121     default:
 122       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, propid, pspec);
 123       break;
 124   }
 125 }
 126
 127 /**
 128  * gst_rtsp_session_pool_new:
 129  *
 130  * Create a new #GstRTSPSessionPool instance.
 131  *
 132  * Returns: A new #GstRTSPSessionPool. g_object_unref() after usage.
 133  */
 134 GstRTSPSessionPool *
 135 gst_rtsp_session_pool_new (void)
 136 {
 137   GstRTSPSessionPool *result;
 138
 139   result = g_object_new (GST_TYPE_RTSP_SESSION_POOL, NULL);
 140
 141   return result;
 142 }
 143
 144 /**
 145  * gst_rtsp_session_pool_set_max_sessions:
 146  * @pool: a #GstRTSPSessionPool
 147  * @max: the maximum number of sessions
 148  *
 149  * Configure the maximum allowed number of sessions in @pool to @max.
 150  * A value of 0 means an unlimited amount of sessions.
 151  */
 152 void
 153 gst_rtsp_session_pool_set_max_sessions (GstRTSPSessionPool * pool, guint max)
 154 {
 155   g_return_if_fail (GST_IS_RTSP_SESSION_POOL (pool));
 156
 157   g_mutex_lock (&pool->lock);
 158   pool->max_sessions = max;
 159   g_mutex_unlock (&pool->lock);
 160 }
 161
 162 /**
 163  * gst_rtsp_session_pool_get_max_sessions:
 164  * @pool: a #GstRTSPSessionPool
 165  *
 166  * Get the maximum allowed number of sessions in @pool. 0 means an unlimited
 167  * amount of sessions.
 168  *
 169  * Returns: the maximum allowed number of sessions.
 170  */
 171 guint
 172 gst_rtsp_session_pool_get_max_sessions (GstRTSPSessionPool * pool)
 173 {
 174   guint result;
 175
 176   g_return_val_if_fail (GST_IS_RTSP_SESSION_POOL (pool), 0);
 177
 178   g_mutex_lock (&pool->lock);
 179   result = pool->max_sessions;
 180   g_mutex_unlock (&pool->lock);
 181
 182   return result;
 183 }
 184
 185 /**
 186  * gst_rtsp_session_pool_get_n_sessions:
 187  * @pool: a #GstRTSPSessionPool
 188  *
 189  * Get the amount of active sessions in @pool.
 190  *
 191  * Returns: the amount of active sessions in @pool.
 192  */
 193 guint
 194 gst_rtsp_session_pool_get_n_sessions (GstRTSPSessionPool * pool)
 195 {
 196   guint result;
 197
 198   g_return_val_if_fail (GST_IS_RTSP_SESSION_POOL (pool), 0);
 199
 200   g_mutex_lock (&pool->lock);
 201   result = g_hash_table_size (pool->sessions);
 202   g_mutex_unlock (&pool->lock);
 203
 204   return result;
 205 }
 206
 207 /**
 208  * gst_rtsp_session_pool_find:
 209  * @pool: the pool to search
 210  * @sessionid: the session id
 211  *
 212  * Find the session with @sessionid in @pool. The access time of the session
 213  * will be updated with gst_rtsp_session_touch().
 214  *
 215  * Returns: (transfer full): the #GstRTSPSession with @sessionid or %NULL when the session did
 216  * not exist. g_object_unref() after usage.
 217  */
 218 GstRTSPSession *
 219 gst_rtsp_session_pool_find (GstRTSPSessionPool * pool, const gchar * sessionid)
 220 {
 221   GstRTSPSession *result;
 222
 223   g_return_val_if_fail (GST_IS_RTSP_SESSION_POOL (pool), NULL);
 224   g_return_val_if_fail (sessionid != NULL, NULL);
 225
 226   g_mutex_lock (&pool->lock);
 227   result = g_hash_table_lookup (pool->sessions, sessionid);
 228   if (result) {
 229     g_object_ref (result);
 230     gst_rtsp_session_touch (result);
 231   }
 232   g_mutex_unlock (&pool->lock);
 233
 234   return result;
 235 }
 236
 237 static gchar *
 238 create_session_id (GstRTSPSessionPool * pool)
 239 {
 240   gchar id[16];
 241   gint i;
 242
 243   for (i = 0; i < 16; i++) {
 244     id[i] =
 245         session_id_charset[g_random_int_range (0,
 246             G_N_ELEMENTS (session_id_charset))];
 247   }
 248
 249   return g_strndup (id, 16);
 250 }
 251
 252 /**
 253  * gst_rtsp_session_pool_create:
 254  * @pool: a #GstRTSPSessionPool
 255  *
 256  * Create a new #GstRTSPSession object in @pool.
 257  *
 258  * Returns: (transfer none): a new #GstRTSPSession.
 259  */
 260 GstRTSPSession *
 261 gst_rtsp_session_pool_create (GstRTSPSessionPool * pool)
 262 {
 263   GstRTSPSession *result = NULL;
 264   GstRTSPSessionPoolClass *klass;
 265   gchar *id = NULL;
 266   guint retry;
 267
 268   g_return_val_if_fail (GST_IS_RTSP_SESSION_POOL (pool), NULL);
 269
 270   klass = GST_RTSP_SESSION_POOL_GET_CLASS (pool);
 271
 272   retry = 0;
 273   do {
 274     /* start by creating a new random session id, we assume that this is random
 275      * enough to not cause a collision, which we will check later  */
 276     if (klass->create_session_id)
 277       id = klass->create_session_id (pool);
 278     else
 279       goto no_function;
 280
 281     if (id == NULL)
 282       goto no_session;
 283
 284     g_mutex_lock (&pool->lock);
 285     /* check session limit */
 286     if (pool->max_sessions > 0) {
 287       if (g_hash_table_size (pool->sessions) >= pool->max_sessions)
 288         goto too_many_sessions;
 289     }
 290     /* check if the sessionid existed */
 291     result = g_hash_table_lookup (pool->sessions, id);
 292     if (result) {
 293       /* found, retry with a different session id */
 294       result = NULL;
 295       retry++;
 296       if (retry > 100)
 297         goto collision;
 298     } else {
 299       /* not found, create session and insert it in the pool */
 300       result = gst_rtsp_session_new (id);
 301       /* take additional ref for the pool */
 302       g_object_ref (result);
 303       g_hash_table_insert (pool->sessions, result->sessionid, result);
 304     }
 305     g_mutex_unlock (&pool->lock);
 306
 307     g_free (id);
 308   } while (result == NULL);
 309
 310   return result;
 311
 312   /* ERRORS */
 313 no_function:
 314   {
 315     GST_WARNING ("no create_session_id vmethod in GstRTSPSessionPool %p", pool);
 316     return NULL;
 317   }
 318 no_session:
 319   {
 320     GST_WARNING ("can't create session id with GstRTSPSessionPool %p", pool);
 321     return NULL;
 322   }
 323 collision:
 324   {
 325     GST_WARNING ("can't find unique sessionid for GstRTSPSessionPool %p", pool);
 326     g_mutex_unlock (&pool->lock);
 327     g_free (id);
 328     return NULL;
 329   }
 330 too_many_sessions:
 331   {
 332     GST_WARNING ("session pool reached max sessions of %d", pool->max_sessions);
 333     g_mutex_unlock (&pool->lock);
 334     g_free (id);
 335     return NULL;
 336   }
 337 }
 338
 339 /**
 340  * gst_rtsp_session_pool_remove:
 341  * @pool: a #GstRTSPSessionPool
 342  * @sess: a #GstRTSPSession
 343  *
 344  * Remove @sess from @pool, releasing the ref that the pool has on @sess.
 345  *
 346  * Returns: %TRUE if the session was found and removed.
 347  */
 348 gboolean
 349 gst_rtsp_session_pool_remove (GstRTSPSessionPool * pool, GstRTSPSession * sess)
 350 {
 351   gboolean found;
 352
 353   g_return_val_if_fail (GST_IS_RTSP_SESSION_POOL (pool), FALSE);
 354   g_return_val_if_fail (GST_IS_RTSP_SESSION (sess), FALSE);
 355
 356   g_mutex_lock (&pool->lock);
 357   found = g_hash_table_remove (pool->sessions, sess->sessionid);
 358   g_mutex_unlock (&pool->lock);
 359
 360   return found;
 361 }
 362
 363 static gboolean
 364 cleanup_func (gchar * sessionid, GstRTSPSession * sess, GTimeVal * now)
 365 {
 366   return gst_rtsp_session_is_expired (sess, now);
 367 }
 368
 369 /**
 370  * gst_rtsp_session_pool_cleanup:
 371  * @pool: a #GstRTSPSessionPool
 372  *
 373  * Inspect all the sessions in @pool and remove the sessions that are inactive
 374  * for more than their timeout.
 375  *
 376  * Returns: the amount of sessions that got removed.
 377  */
 378 guint
 379 gst_rtsp_session_pool_cleanup (GstRTSPSessionPool * pool)
 380 {
 381   guint result;
 382   GTimeVal now;
 383
 384   g_return_val_if_fail (GST_IS_RTSP_SESSION_POOL (pool), 0);
 385
 386   g_get_current_time (&now);
 387
 388   g_mutex_lock (&pool->lock);
 389   result =
 390       g_hash_table_foreach_remove (pool->sessions, (GHRFunc) cleanup_func,
 391       &now);
 392   g_mutex_unlock (&pool->lock);
 393
 394   return result;
 395 }
 396
 397 typedef struct
 398 {
 399   GstRTSPSessionPool *pool;
 400   GstRTSPSessionFilterFunc func;
 401   gpointer user_data;
 402   GList *list;
 403 } FilterData;
 404
 405 static gboolean
 406 filter_func (gchar * sessionid, GstRTSPSession * sess, FilterData * data)
 407 {
 408   switch (data->func (data->pool, sess, data->user_data)) {
 409     case GST_RTSP_FILTER_REMOVE:
 410       return TRUE;
 411     case GST_RTSP_FILTER_REF:
 412       /* keep ref */
 413       data->list = g_list_prepend (data->list, g_object_ref (sess));
 414       /* fallthrough */
 415     default:
 416     case GST_RTSP_FILTER_KEEP:
 417       return FALSE;
 418   }
 419 }
 420
 421 /**
 422  * gst_rtsp_session_pool_filter:
 423  * @pool: a #GstRTSPSessionPool
 424  * @func: (scope call): a callback
 425  * @user_data: user data passed to @func
 426  *
 427  * Call @func for each session in @pool. The result value of @func determines
 428  * what happens to the session. @func will be called with the session pool
 429  * locked so no further actions on @pool can be performed from @func.
 430  *
 431  * If @func returns #GST_RTSP_FILTER_REMOVE, the session will be removed from
 432  * @pool.
 433  *
 434  * If @func returns #GST_RTSP_FILTER_KEEP, the session will remain in @pool.
 435  *
 436  * If @func returns #GST_RTSP_FILTER_REF, the session will remain in @pool but
 437  * will also be added with an additional ref to the result GList of this
 438  * function..
 439  *
 440  * Returns: (element-type GstRTSPSession) (transfer full): a GList with all
 441  * sessions for which @func returned #GST_RTSP_FILTER_REF. After usage, each
 442  * element in the GList should be unreffed before the list is freed.
 443  */
 444 GList *
 445 gst_rtsp_session_pool_filter (GstRTSPSessionPool * pool,
 446     GstRTSPSessionFilterFunc func, gpointer user_data)
 447 {
 448   FilterData data;
 449
 450   g_return_val_if_fail (GST_IS_RTSP_SESSION_POOL (pool), NULL);
 451   g_return_val_if_fail (func != NULL, NULL);
 452
 453   data.pool = pool;
 454   data.func = func;
 455   data.user_data = user_data;
 456   data.list = NULL;
 457
 458   g_mutex_lock (&pool->lock);
 459   g_hash_table_foreach_remove (pool->sessions, (GHRFunc) filter_func, &data);
 460   g_mutex_unlock (&pool->lock);
 461
 462   return data.list;
 463 }
 464
 465 typedef struct
 466 {
 467   GSource source;
 468   GstRTSPSessionPool *pool;
 469   gint timeout;
 470 } GstPoolSource;
 471
 472 static void
 473 collect_timeout (gchar * sessionid, GstRTSPSession * sess, GstPoolSource * psrc)
 474 {
 475   gint timeout;
 476   GTimeVal now;
 477   gint64 tmp;
 478
 479   tmp = g_source_get_time ((GSource *) psrc);
 480   now.tv_sec = tmp / G_USEC_PER_SEC;
 481   now.tv_usec = tmp % G_USEC_PER_SEC;
 482
 483   timeout = gst_rtsp_session_next_timeout (sess, &now);
 484   GST_INFO ("%p: next timeout: %d", sess, timeout);
 485   if (psrc->timeout == -1 || timeout < psrc->timeout)
 486     psrc->timeout = timeout;
 487 }
 488
 489 static gboolean
 490 gst_pool_source_prepare (GSource * source, gint * timeout)
 491 {
 492   GstPoolSource *psrc;
 493   gboolean result;
 494
 495   psrc = (GstPoolSource *) source;
 496   psrc->timeout = -1;
 497
 498   g_mutex_lock (&psrc->pool->lock);
 499   g_hash_table_foreach (psrc->pool->sessions, (GHFunc) collect_timeout, psrc);
 500   g_mutex_unlock (&psrc->pool->lock);
 501
 502   if (timeout)
 503     *timeout = psrc->timeout;
 504
 505   result = psrc->timeout == 0;
 506
 507   GST_INFO ("prepare %d, %d", psrc->timeout, result);
 508
 509   return result;
 510 }
 511
 512 static gboolean
 513 gst_pool_source_check (GSource * source)
 514 {
 515   GST_INFO ("check");
 516
 517   return gst_pool_source_prepare (source, NULL);
 518 }
 519
 520 static gboolean
 521 gst_pool_source_dispatch (GSource * source, GSourceFunc callback,
 522     gpointer user_data)
 523 {
 524   gboolean res;
 525   GstPoolSource *psrc = (GstPoolSource *) source;
 526   GstRTSPSessionPoolFunc func = (GstRTSPSessionPoolFunc) callback;
 527
 528   GST_INFO ("dispatch");
 529
 530   if (func)
 531     res = func (psrc->pool, user_data);
 532   else
 533     res = FALSE;
 534
 535   return res;
 536 }
 537
 538 static void
 539 gst_pool_source_finalize (GSource * source)
 540 {
 541   GstPoolSource *psrc = (GstPoolSource *) source;
 542
 543   GST_INFO ("finalize %p", psrc);
 544
 545   g_object_unref (psrc->pool);
 546   psrc->pool = NULL;
 547 }
 548
 549 static GSourceFuncs gst_pool_source_funcs = {
 550   gst_pool_source_prepare,
 551   gst_pool_source_check,
 552   gst_pool_source_dispatch,
 553   gst_pool_source_finalize
 554 };
 555
 556 /**
 557  * gst_rtsp_session_pool_create_watch:
 558  * @pool: a #GstRTSPSessionPool
 559  *
 560  * A GSource that will be dispatched when the session should be cleaned up.
 561  */
 562 GSource *
 563 gst_rtsp_session_pool_create_watch (GstRTSPSessionPool * pool)
 564 {
 565   GstPoolSource *source;
 566
 567   g_return_val_if_fail (GST_IS_RTSP_SESSION_POOL (pool), NULL);
 568
 569   source = (GstPoolSource *) g_source_new (&gst_pool_source_funcs,
 570       sizeof (GstPoolSource));
 571   source->pool = g_object_ref (pool);
 572
 573   return (GSource *) source;
 574 }