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  * SECTION:rtsp-session-pool
  21  * @short_description: An object for managing sessions
  22  * @see_also: #GstRTSPSession
  23  *
  24  * The #GstRTSPSessionPool object manages a list of #GstRTSPSession objects.
  25  *
  26  * The maximum number of sessions can be configured with
  27  * gst_rtsp_session_pool_set_max_sessions(). The current number of sessions can
  28  * be retrieved with gst_rtsp_session_pool_get_n_sessions().
  29  *
  30  * Use gst_rtsp_session_pool_create() to create a new #GstRTSPSession object.
  31  * The session object can be found again with its id and
  32  * gst_rtsp_session_pool_find().
  33  *
  34  * All sessions can be iterated with gst_rtsp_session_pool_filter().
  35  *
  36  * Run gst_rtsp_session_pool_cleanup() periodically to remove timed out sessions
  37  * or use gst_rtsp_session_pool_create_watch() to be notified when session
  38  * cleanup should be performed.
  39  *
  40  * Last reviewed on 2013-07-11 (1.0.0)
  41  */
  42
  43 #include "rtsp-session-pool.h"
  44
  45 #define GST_RTSP_SESSION_POOL_GET_PRIVATE(obj)  \
  46          (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_RTSP_SESSION_POOL, GstRTSPSessionPoolPrivate))
  47
  48 struct _GstRTSPSessionPoolPrivate
  49 {
  50   GMutex lock;                  /* protects everything in this struct */
  51   guint max_sessions;
  52   GHashTable *sessions;
  53   guint sessions_cookie;
  54 };
  55
  56 #define DEFAULT_MAX_SESSIONS 0
  57
  58 enum
  59 {
  60   PROP_0,
  61   PROP_MAX_SESSIONS,
  62   PROP_LAST
  63 };
  64
  65 static const gchar session_id_charset[] =
  66     { 'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o',
  67   'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z', 'A', 'B', 'C', 'D',
  68   'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', 'N', 'O', 'P', 'Q', 'R', 'S',
  69   'T', 'U', 'V', 'W', 'X', 'Y', 'Z', '0', '1', '2', '3', '4', '5', '6', '7',
  70   '8', '9', '$', '-', '_', '.', '+'
  71 };
  72
  73 enum
  74 {
  75   SIGNAL_SESSION_REMOVED,
  76   SIGNAL_LAST
  77 };
  78
  79 static guint gst_rtsp_session_pool_signals[SIGNAL_LAST] = { 0 };
  80
  81 GST_DEBUG_CATEGORY_STATIC (rtsp_session_debug);
  82 #define GST_CAT_DEFAULT rtsp_session_debug
  83
  84 static void gst_rtsp_session_pool_get_property (GObject * object, guint propid,
  85     GValue * value, GParamSpec * pspec);
  86 static void gst_rtsp_session_pool_set_property (GObject * object, guint propid,
  87     const GValue * value, GParamSpec * pspec);
  88 static void gst_rtsp_session_pool_finalize (GObject * object);
  89
  90 static gchar *create_session_id (GstRTSPSessionPool * pool);
  91 static GstRTSPSession *create_session (GstRTSPSessionPool * pool,
  92     const gchar * id);
  93
  94 G_DEFINE_TYPE (GstRTSPSessionPool, gst_rtsp_session_pool, G_TYPE_OBJECT);
  95
  96 static void
  97 gst_rtsp_session_pool_class_init (GstRTSPSessionPoolClass * klass)
  98 {
  99   GObjectClass *gobject_class;
 100
 101   g_type_class_add_private (klass, sizeof (GstRTSPSessionPoolPrivate));
 102
 103   gobject_class = G_OBJECT_CLASS (klass);
 104
 105   gobject_class->get_property = gst_rtsp_session_pool_get_property;
 106   gobject_class->set_property = gst_rtsp_session_pool_set_property;
 107   gobject_class->finalize = gst_rtsp_session_pool_finalize;
 108
 109   g_object_class_install_property (gobject_class, PROP_MAX_SESSIONS,
 110       g_param_spec_uint ("max-sessions", "Max Sessions",
 111           "the maximum amount of sessions (0 = unlimited)",
 112           0, G_MAXUINT, DEFAULT_MAX_SESSIONS,
 113           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 114
 115   gst_rtsp_session_pool_signals[SIGNAL_SESSION_REMOVED] =
 116       g_signal_new ("session-removed", G_TYPE_FROM_CLASS (klass),
 117       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRTSPSessionPoolClass,
 118           session_removed), NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE,
 119       1, GST_TYPE_RTSP_SESSION);
 120
 121   klass->create_session_id = create_session_id;
 122   klass->create_session = create_session;
 123
 124   GST_DEBUG_CATEGORY_INIT (rtsp_session_debug, "rtspsessionpool", 0,
 125       "GstRTSPSessionPool");
 126 }
 127
 128 static void
 129 gst_rtsp_session_pool_init (GstRTSPSessionPool * pool)
 130 {
 131   GstRTSPSessionPoolPrivate *priv = GST_RTSP_SESSION_POOL_GET_PRIVATE (pool);
 132
 133   pool->priv = priv;
 134
 135   g_mutex_init (&priv->lock);
 136   priv->sessions = g_hash_table_new_full (g_str_hash, g_str_equal,
 137       NULL, g_object_unref);
 138   priv->max_sessions = DEFAULT_MAX_SESSIONS;
 139 }
 140
 141 static GstRTSPFilterResult
 142 remove_sessions_func (GstRTSPSessionPool * pool, GstRTSPSession * session,
 143     gpointer user_data)
 144 {
 145   return GST_RTSP_FILTER_REMOVE;
 146 }
 147
 148 static void
 149 gst_rtsp_session_pool_finalize (GObject * object)
 150 {
 151   GstRTSPSessionPool *pool = GST_RTSP_SESSION_POOL (object);
 152   GstRTSPSessionPoolPrivate *priv = pool->priv;
 153
 154   gst_rtsp_session_pool_filter (pool, remove_sessions_func, NULL);
 155   g_hash_table_unref (priv->sessions);
 156   g_mutex_clear (&priv->lock);
 157
 158   G_OBJECT_CLASS (gst_rtsp_session_pool_parent_class)->finalize (object);
 159 }
 160
 161 static void
 162 gst_rtsp_session_pool_get_property (GObject * object, guint propid,
 163     GValue * value, GParamSpec * pspec)
 164 {
 165   GstRTSPSessionPool *pool = GST_RTSP_SESSION_POOL (object);
 166
 167   switch (propid) {
 168     case PROP_MAX_SESSIONS:
 169       g_value_set_uint (value, gst_rtsp_session_pool_get_max_sessions (pool));
 170       break;
 171     default:
 172       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, propid, pspec);
 173       break;
 174   }
 175 }
 176
 177 static void
 178 gst_rtsp_session_pool_set_property (GObject * object, guint propid,
 179     const GValue * value, GParamSpec * pspec)
 180 {
 181   GstRTSPSessionPool *pool = GST_RTSP_SESSION_POOL (object);
 182
 183   switch (propid) {
 184     case PROP_MAX_SESSIONS:
 185       gst_rtsp_session_pool_set_max_sessions (pool, g_value_get_uint (value));
 186       break;
 187     default:
 188       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, propid, pspec);
 189       break;
 190   }
 191 }
 192
 193 /**
 194  * gst_rtsp_session_pool_new:
 195  *
 196  * Create a new #GstRTSPSessionPool instance.
 197  *
 198  * Returns: (transfer full): A new #GstRTSPSessionPool. g_object_unref() after
 199  * usage.
 200  */
 201 GstRTSPSessionPool *
 202 gst_rtsp_session_pool_new (void)
 203 {
 204   GstRTSPSessionPool *result;
 205
 206   result = g_object_new (GST_TYPE_RTSP_SESSION_POOL, NULL);
 207
 208   return result;
 209 }
 210
 211 /**
 212  * gst_rtsp_session_pool_set_max_sessions:
 213  * @pool: a #GstRTSPSessionPool
 214  * @max: the maximum number of sessions
 215  *
 216  * Configure the maximum allowed number of sessions in @pool to @max.
 217  * A value of 0 means an unlimited amount of sessions.
 218  */
 219 void
 220 gst_rtsp_session_pool_set_max_sessions (GstRTSPSessionPool * pool, guint max)
 221 {
 222   GstRTSPSessionPoolPrivate *priv;
 223
 224   g_return_if_fail (GST_IS_RTSP_SESSION_POOL (pool));
 225
 226   priv = pool->priv;
 227
 228   g_mutex_lock (&priv->lock);
 229   priv->max_sessions = max;
 230   g_mutex_unlock (&priv->lock);
 231 }
 232
 233 /**
 234  * gst_rtsp_session_pool_get_max_sessions:
 235  * @pool: a #GstRTSPSessionPool
 236  *
 237  * Get the maximum allowed number of sessions in @pool. 0 means an unlimited
 238  * amount of sessions.
 239  *
 240  * Returns: the maximum allowed number of sessions.
 241  */
 242 guint
 243 gst_rtsp_session_pool_get_max_sessions (GstRTSPSessionPool * pool)
 244 {
 245   GstRTSPSessionPoolPrivate *priv;
 246   guint result;
 247
 248   g_return_val_if_fail (GST_IS_RTSP_SESSION_POOL (pool), 0);
 249
 250   priv = pool->priv;
 251
 252   g_mutex_lock (&priv->lock);
 253   result = priv->max_sessions;
 254   g_mutex_unlock (&priv->lock);
 255
 256   return result;
 257 }
 258
 259 /**
 260  * gst_rtsp_session_pool_get_n_sessions:
 261  * @pool: a #GstRTSPSessionPool
 262  *
 263  * Get the amount of active sessions in @pool.
 264  *
 265  * Returns: the amount of active sessions in @pool.
 266  */
 267 guint
 268 gst_rtsp_session_pool_get_n_sessions (GstRTSPSessionPool * pool)
 269 {
 270   GstRTSPSessionPoolPrivate *priv;
 271   guint result;
 272
 273   g_return_val_if_fail (GST_IS_RTSP_SESSION_POOL (pool), 0);
 274
 275   priv = pool->priv;
 276
 277   g_mutex_lock (&priv->lock);
 278   result = g_hash_table_size (priv->sessions);
 279   g_mutex_unlock (&priv->lock);
 280
 281   return result;
 282 }
 283
 284 /**
 285  * gst_rtsp_session_pool_find:
 286  * @pool: the pool to search
 287  * @sessionid: the session id
 288  *
 289  * Find the session with @sessionid in @pool. The access time of the session
 290  * will be updated with gst_rtsp_session_touch().
 291  *
 292  * Returns: (transfer full) (nullable): the #GstRTSPSession with @sessionid
 293  * or %NULL when the session did not exist. g_object_unref() after usage.
 294  */
 295 GstRTSPSession *
 296 gst_rtsp_session_pool_find (GstRTSPSessionPool * pool, const gchar * sessionid)
 297 {
 298   GstRTSPSessionPoolPrivate *priv;
 299   GstRTSPSession *result;
 300
 301   g_return_val_if_fail (GST_IS_RTSP_SESSION_POOL (pool), NULL);
 302   g_return_val_if_fail (sessionid != NULL, NULL);
 303
 304   priv = pool->priv;
 305
 306   g_mutex_lock (&priv->lock);
 307   result = g_hash_table_lookup (priv->sessions, sessionid);
 308   if (result) {
 309     g_object_ref (result);
 310     gst_rtsp_session_touch (result);
 311   }
 312   g_mutex_unlock (&priv->lock);
 313
 314   return result;
 315 }
 316
 317 static gchar *
 318 create_session_id (GstRTSPSessionPool * pool)
 319 {
 320   gchar id[17];
 321   gint i;
 322
 323   for (i = 0; i < 16; i++) {
 324     id[i] =
 325         session_id_charset[g_random_int_range (0,
 326             G_N_ELEMENTS (session_id_charset))];
 327   }
 328   id[16] = 0;
 329
 330   return g_uri_escape_string (id, NULL, FALSE);
 331 }
 332
 333 static GstRTSPSession *
 334 create_session (GstRTSPSessionPool * pool, const gchar * id)
 335 {
 336   return gst_rtsp_session_new (id);
 337 }
 338
 339 /**
 340  * gst_rtsp_session_pool_create:
 341  * @pool: a #GstRTSPSessionPool
 342  *
 343  * Create a new #GstRTSPSession object in @pool.
 344  *
 345  * Returns: (transfer full): a new #GstRTSPSession.
 346  */
 347 GstRTSPSession *
 348 gst_rtsp_session_pool_create (GstRTSPSessionPool * pool)
 349 {
 350   GstRTSPSessionPoolPrivate *priv;
 351   GstRTSPSession *result = NULL;
 352   GstRTSPSessionPoolClass *klass;
 353   gchar *id = NULL;
 354   guint retry;
 355
 356   g_return_val_if_fail (GST_IS_RTSP_SESSION_POOL (pool), NULL);
 357
 358   priv = pool->priv;
 359
 360   klass = GST_RTSP_SESSION_POOL_GET_CLASS (pool);
 361
 362   retry = 0;
 363   do {
 364     /* start by creating a new random session id, we assume that this is random
 365      * enough to not cause a collision, which we will check later  */
 366     if (klass->create_session_id)
 367       id = klass->create_session_id (pool);
 368     else
 369       goto no_function;
 370
 371     if (id == NULL)
 372       goto no_session;
 373
 374     g_mutex_lock (&priv->lock);
 375     /* check session limit */
 376     if (priv->max_sessions > 0) {
 377       if (g_hash_table_size (priv->sessions) >= priv->max_sessions)
 378         goto too_many_sessions;
 379     }
 380     /* check if the sessionid existed */
 381     result = g_hash_table_lookup (priv->sessions, id);
 382     if (result) {
 383       /* found, retry with a different session id */
 384       result = NULL;
 385       retry++;
 386       if (retry > 100)
 387         goto collision;
 388     } else {
 389       /* not found, create session and insert it in the pool */
 390       if (klass->create_session)
 391         result = create_session (pool, id);
 392       if (result == NULL)
 393         goto too_many_sessions;
 394       /* take additional ref for the pool */
 395       g_object_ref (result);
 396       g_hash_table_insert (priv->sessions,
 397           (gchar *) gst_rtsp_session_get_sessionid (result), result);
 398       priv->sessions_cookie++;
 399     }
 400     g_mutex_unlock (&priv->lock);
 401
 402     g_free (id);
 403   } while (result == NULL);
 404
 405   return result;
 406
 407   /* ERRORS */
 408 no_function:
 409   {
 410     GST_WARNING ("no create_session_id vmethod in GstRTSPSessionPool %p", pool);
 411     return NULL;
 412   }
 413 no_session:
 414   {
 415     GST_WARNING ("can't create session id with GstRTSPSessionPool %p", pool);
 416     return NULL;
 417   }
 418 collision:
 419   {
 420     GST_WARNING ("can't find unique sessionid for GstRTSPSessionPool %p", pool);
 421     g_mutex_unlock (&priv->lock);
 422     g_free (id);
 423     return NULL;
 424   }
 425 too_many_sessions:
 426   {
 427     GST_WARNING ("session pool reached max sessions of %d", priv->max_sessions);
 428     g_mutex_unlock (&priv->lock);
 429     g_free (id);
 430     return NULL;
 431   }
 432 }
 433
 434 /**
 435  * gst_rtsp_session_pool_remove:
 436  * @pool: a #GstRTSPSessionPool
 437  * @sess: (transfer none): a #GstRTSPSession
 438  *
 439  * Remove @sess from @pool, releasing the ref that the pool has on @sess.
 440  *
 441  * Returns: %TRUE if the session was found and removed.
 442  */
 443 gboolean
 444 gst_rtsp_session_pool_remove (GstRTSPSessionPool * pool, GstRTSPSession * sess)
 445 {
 446   GstRTSPSessionPoolPrivate *priv;
 447   gboolean found;
 448
 449   g_return_val_if_fail (GST_IS_RTSP_SESSION_POOL (pool), FALSE);
 450   g_return_val_if_fail (GST_IS_RTSP_SESSION (sess), FALSE);
 451
 452   priv = pool->priv;
 453
 454   g_mutex_lock (&priv->lock);
 455   g_object_ref (sess);
 456   found =
 457       g_hash_table_remove (priv->sessions,
 458       gst_rtsp_session_get_sessionid (sess));
 459   if (found)
 460     priv->sessions_cookie++;
 461   g_mutex_unlock (&priv->lock);
 462
 463   if (found)
 464     g_signal_emit (pool, gst_rtsp_session_pool_signals[SIGNAL_SESSION_REMOVED],
 465         0, sess);
 466
 467   g_object_unref (sess);
 468
 469   return found;
 470 }
 471
 472 typedef struct
 473 {
 474   GTimeVal now;
 475   GstRTSPSessionPool *pool;
 476   GList *removed;
 477 } CleanupData;
 478
 479 static gboolean
 480 cleanup_func (gchar * sessionid, GstRTSPSession * sess, CleanupData * data)
 481 {
 482   gboolean expired;
 483
 484   expired = gst_rtsp_session_is_expired (sess, &data->now);
 485   if (expired) {
 486     GST_DEBUG ("session expired");
 487     data->removed = g_list_prepend (data->removed, g_object_ref (sess));
 488   }
 489   return expired;
 490 }
 491
 492 /**
 493  * gst_rtsp_session_pool_cleanup:
 494  * @pool: a #GstRTSPSessionPool
 495  *
 496  * Inspect all the sessions in @pool and remove the sessions that are inactive
 497  * for more than their timeout.
 498  *
 499  * Returns: the amount of sessions that got removed.
 500  */
 501 guint
 502 gst_rtsp_session_pool_cleanup (GstRTSPSessionPool * pool)
 503 {
 504   GstRTSPSessionPoolPrivate *priv;
 505   guint result;
 506   CleanupData data;
 507   GList *walk;
 508
 509   g_return_val_if_fail (GST_IS_RTSP_SESSION_POOL (pool), 0);
 510
 511   priv = pool->priv;
 512
 513   g_get_current_time (&data.now);
 514   data.pool = pool;
 515   data.removed = NULL;
 516
 517   g_mutex_lock (&priv->lock);
 518   result =
 519       g_hash_table_foreach_remove (priv->sessions, (GHRFunc) cleanup_func,
 520       &data);
 521   if (result > 0)
 522     priv->sessions_cookie++;
 523   g_mutex_unlock (&priv->lock);
 524
 525   for (walk = data.removed; walk; walk = walk->next) {
 526     GstRTSPSession *sess = walk->data;
 527
 528     g_signal_emit (pool,
 529         gst_rtsp_session_pool_signals[SIGNAL_SESSION_REMOVED], 0, sess);
 530
 531     g_object_unref (sess);
 532   }
 533   g_list_free (data.removed);
 534
 535   return result;
 536 }
 537
 538 /**
 539  * gst_rtsp_session_pool_filter:
 540  * @pool: a #GstRTSPSessionPool
 541  * @func: (scope call) (allow-none): a callback
 542  * @user_data: (closure): user data passed to @func
 543  *
 544  * Call @func for each session in @pool. The result value of @func determines
 545  * what happens to the session. @func will be called with the session pool
 546  * locked so no further actions on @pool can be performed from @func.
 547  *
 548  * If @func returns #GST_RTSP_FILTER_REMOVE, the session will be set to the
 549  * expired state with gst_rtsp_session_set_expired() and removed from
 550  * @pool.
 551  *
 552  * If @func returns #GST_RTSP_FILTER_KEEP, the session will remain in @pool.
 553  *
 554  * If @func returns #GST_RTSP_FILTER_REF, the session will remain in @pool but
 555  * will also be added with an additional ref to the result GList of this
 556  * function..
 557  *
 558  * When @func is %NULL, #GST_RTSP_FILTER_REF will be assumed for all sessions.
 559  *
 560  * Returns: (element-type GstRTSPSession) (transfer full): a GList with all
 561  * sessions for which @func returned #GST_RTSP_FILTER_REF. After usage, each
 562  * element in the GList should be unreffed before the list is freed.
 563  */
 564 GList *
 565 gst_rtsp_session_pool_filter (GstRTSPSessionPool * pool,
 566     GstRTSPSessionPoolFilterFunc func, gpointer user_data)
 567 {
 568   GstRTSPSessionPoolPrivate *priv;
 569   GHashTableIter iter;
 570   gpointer key, value;
 571   GList *result;
 572   GHashTable *visited;
 573   guint cookie;
 574
 575   g_return_val_if_fail (GST_IS_RTSP_SESSION_POOL (pool), NULL);
 576
 577   priv = pool->priv;
 578
 579   result = NULL;
 580   if (func)
 581     visited = g_hash_table_new_full (NULL, NULL, g_object_unref, NULL);
 582
 583   g_mutex_lock (&priv->lock);
 584 restart:
 585   g_hash_table_iter_init (&iter, priv->sessions);
 586   cookie = priv->sessions_cookie;
 587   while (g_hash_table_iter_next (&iter, &key, &value)) {
 588     GstRTSPSession *session = value;
 589     GstRTSPFilterResult res;
 590     gboolean changed;
 591
 592     if (func) {
 593       /* only visit each session once */
 594       if (g_hash_table_contains (visited, session))
 595         continue;
 596
 597       g_hash_table_add (visited, g_object_ref (session));
 598       g_mutex_unlock (&priv->lock);
 599
 600       res = func (pool, session, user_data);
 601
 602       g_mutex_lock (&priv->lock);
 603     } else
 604       res = GST_RTSP_FILTER_REF;
 605
 606     changed = (cookie != priv->sessions_cookie);
 607
 608     switch (res) {
 609       case GST_RTSP_FILTER_REMOVE:
 610       {
 611         gboolean removed = TRUE;
 612
 613         if (changed)
 614           /* something changed, check if we still have the session */
 615           removed = g_hash_table_remove (priv->sessions, key);
 616         else
 617           g_hash_table_iter_remove (&iter);
 618
 619         if (removed) {
 620           /* if we managed to remove the session, update the cookie and
 621            * signal */
 622           cookie = ++priv->sessions_cookie;
 623           g_mutex_unlock (&priv->lock);
 624
 625           g_signal_emit (pool,
 626               gst_rtsp_session_pool_signals[SIGNAL_SESSION_REMOVED], 0,
 627               session);
 628
 629           g_mutex_lock (&priv->lock);
 630           /* cookie could have changed again, make sure we restart */
 631           changed |= (cookie != priv->sessions_cookie);
 632         }
 633         break;
 634       }
 635       case GST_RTSP_FILTER_REF:
 636         /* keep ref */
 637         result = g_list_prepend (result, g_object_ref (session));
 638         break;
 639       case GST_RTSP_FILTER_KEEP:
 640       default:
 641         break;
 642     }
 643     if (changed)
 644       goto restart;
 645   }
 646   g_mutex_unlock (&priv->lock);
 647
 648   if (func)
 649     g_hash_table_unref (visited);
 650
 651   return result;
 652 }
 653
 654 typedef struct
 655 {
 656   GSource source;
 657   GstRTSPSessionPool *pool;
 658   gint timeout;
 659 } GstPoolSource;
 660
 661 static void
 662 collect_timeout (gchar * sessionid, GstRTSPSession * sess, GstPoolSource * psrc)
 663 {
 664   gint timeout;
 665   GTimeVal now;
 666
 667   g_get_current_time (&now);
 668
 669   timeout = gst_rtsp_session_next_timeout (sess, &now);
 670   GST_INFO ("%p: next timeout: %d", sess, timeout);
 671   if (psrc->timeout == -1 || timeout < psrc->timeout)
 672     psrc->timeout = timeout;
 673 }
 674
 675 static gboolean
 676 gst_pool_source_prepare (GSource * source, gint * timeout)
 677 {
 678   GstRTSPSessionPoolPrivate *priv;
 679   GstPoolSource *psrc;
 680   gboolean result;
 681
 682   psrc = (GstPoolSource *) source;
 683   psrc->timeout = -1;
 684   priv = psrc->pool->priv;
 685
 686   g_mutex_lock (&priv->lock);
 687   g_hash_table_foreach (priv->sessions, (GHFunc) collect_timeout, psrc);
 688   g_mutex_unlock (&priv->lock);
 689
 690   if (timeout)
 691     *timeout = psrc->timeout;
 692
 693   result = psrc->timeout == 0;
 694
 695   GST_INFO ("prepare %d, %d", psrc->timeout, result);
 696
 697   return result;
 698 }
 699
 700 static gboolean
 701 gst_pool_source_check (GSource * source)
 702 {
 703   GST_INFO ("check");
 704
 705   return gst_pool_source_prepare (source, NULL);
 706 }
 707
 708 static gboolean
 709 gst_pool_source_dispatch (GSource * source, GSourceFunc callback,
 710     gpointer user_data)
 711 {
 712   gboolean res;
 713   GstPoolSource *psrc = (GstPoolSource *) source;
 714   GstRTSPSessionPoolFunc func = (GstRTSPSessionPoolFunc) callback;
 715
 716   GST_INFO ("dispatch");
 717
 718   if (func)
 719     res = func (psrc->pool, user_data);
 720   else
 721     res = FALSE;
 722
 723   return res;
 724 }
 725
 726 static void
 727 gst_pool_source_finalize (GSource * source)
 728 {
 729   GstPoolSource *psrc = (GstPoolSource *) source;
 730
 731   GST_INFO ("finalize %p", psrc);
 732
 733   g_object_unref (psrc->pool);
 734   psrc->pool = NULL;
 735 }
 736
 737 static GSourceFuncs gst_pool_source_funcs = {
 738   gst_pool_source_prepare,
 739   gst_pool_source_check,
 740   gst_pool_source_dispatch,
 741   gst_pool_source_finalize
 742 };
 743
 744 /**
 745  * gst_rtsp_session_pool_create_watch:
 746  * @pool: a #GstRTSPSessionPool
 747  *
 748  * Create a #GSource that will be dispatched when the session should be cleaned
 749  * up.
 750  *
 751  * Returns: (transfer full): a #GSource
 752  */
 753 GSource *
 754 gst_rtsp_session_pool_create_watch (GstRTSPSessionPool * pool)
 755 {
 756   GstPoolSource *source;
 757
 758   g_return_val_if_fail (GST_IS_RTSP_SESSION_POOL (pool), NULL);
 759
 760   source = (GstPoolSource *) g_source_new (&gst_pool_source_funcs,
 761       sizeof (GstPoolSource));
 762   source->pool = g_object_ref (pool);
 763
 764   return (GSource *) source;
 765 }