/src/gpac/src/utils/downloader_hmux.c

Source (jump to first uncovered line)
/*
 *          GPAC Multimedia Framework
 *
 *      Authors: Jean Le Feuvre
 *      Copyright (c) Telecom ParisTech 2005-2025
 *          All rights reserved
 *
 *  This file is part of GPAC / downloader sub-project
 *
 *  GPAC is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU Lesser General Public License as published by
 *  the Free Software Foundation; either version 2, or (at your option)
 *  any later version.
 *
 *  GPAC is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU Lesser General Public License for more details.
 *
 *  You should have received a copy of the GNU Lesser General Public
 *  License along with this library; see the file COPYING.  If not, write to
 *  the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
 *
 */

#include "downloader.h"

#ifndef GPAC_DISABLE_NETWORK

#ifdef GPAC_HTTPMUX

//detach session from parent h2 / h3 session - the session mutex SHALL be grabbed before calling this
void hmux_detach_session(GF_HMUX_Session *hmux_sess, GF_DownloadSession *sess)
{
  if (!hmux_sess || !sess) return;
  gf_assert(sess->hmux_sess == hmux_sess);
  gf_assert(sess->mx);

  gf_list_del_item(hmux_sess->sessions, sess);
  if (!gf_list_count(hmux_sess->sessions)) {
    if (hmux_sess->close)
      hmux_sess->close(hmux_sess);

#ifdef GPAC_HAS_SSL
    if (sess->ssl) {
      GF_LOG(GF_LOG_DEBUG, GF_LOG_HTTP, ("[Downloader] shut down SSL context\n"));
      SSL_shutdown(sess->ssl);
      SSL_free(sess->ssl);
      sess->ssl = NULL;
    }
#endif
    dm_sess_sk_del(sess);
    if (hmux_sess->destroy)
      hmux_sess->destroy(hmux_sess);

    gf_list_del(hmux_sess->sessions);
    gf_mx_v(hmux_sess->mx);
    gf_mx_del(sess->mx);
    gf_free(hmux_sess);
  } else {
    GF_DownloadSession *asess = gf_list_get(hmux_sess->sessions, 0);
    gf_assert(asess->hmux_sess == hmux_sess);

    hmux_sess->net_sess = asess;
    //swap async buf if any to new active session
    if (sess->async_buf_alloc) {
      gf_assert(!asess->async_buf);
      asess->async_buf = sess->async_buf;
      asess->async_buf_alloc = sess->async_buf_alloc;
      asess->async_buf_size = sess->async_buf_size;
      sess->async_buf_alloc = sess->async_buf_size = 0;
      sess->async_buf = NULL;
    }
    sess->sock = NULL;
#ifdef GPAC_HAS_SSL
    sess->ssl = NULL;
#endif
    gf_mx_v(sess->mx);
  }

  if (sess->hmux_buf.data) {
    gf_free(sess->hmux_buf.data);
    memset(&sess->hmux_buf, 0, sizeof(hmux_reagg_buffer));
  }
  sess->hmux_sess = NULL;
  sess->mx = NULL;
}


GF_DownloadSession *hmux_get_session(void *user_data, s64 stream_id, Bool can_reassign)
{
  u32 i, nb_sess;
  GF_DownloadSession *first_not_assigned = NULL;
  GF_HMUX_Session *hmux_sess = (GF_HMUX_Session *)user_data;

  nb_sess = gf_list_count(hmux_sess->sessions);
  for (i=0;i<nb_sess; i++) {
    GF_DownloadSession *s = gf_list_get(hmux_sess->sessions, i);
    if (s->hmux_stream_id == stream_id)
      return s;

    if (s->server_mode && (s->hmux_stream_id<0) && !first_not_assigned) {
      first_not_assigned = s;
    }
  }
  if (can_reassign && first_not_assigned) {
    first_not_assigned->hmux_stream_id = stream_id;
    GF_LOG(GF_LOG_DEBUG, GF_LOG_HTTP, ("[%s] reassigning old server session to new stream %d\n", hmux_sess->net_sess->log_name, stream_id));
    first_not_assigned->status = GF_NETIO_CONNECTED;
    first_not_assigned->total_size = first_not_assigned->bytes_done = 0;
    first_not_assigned->hmux_data_done = 0;
    if (first_not_assigned->remote_path) gf_free(first_not_assigned->remote_path);
    first_not_assigned->remote_path = NULL;
    //reset internal reaggregation buffer
    first_not_assigned->hmux_buf.size = 0;

    gf_dm_sess_clear_headers(first_not_assigned);
    return first_not_assigned;
  }
  return NULL;
}

void hmux_flush_internal_data(GF_DownloadSession *sess, Bool store_in_init)
{
  gf_dm_data_received(sess, (u8 *) sess->hmux_buf.data, sess->hmux_buf.size, store_in_init, NULL, NULL);
  sess->hmux_buf.size = 0;
}

void hmux_fetch_data(GF_DownloadSession *sess, u8 *obuffer, u32 size, u32 *nb_bytes)
{
  u32 copy, nb_b_pck;
  u8 *data;
  *nb_bytes = 0;
  if (!sess->hmux_buf.size) {
    if (sess->hmux_is_eos) {
      if (!sess->total_size) {
        sess->total_size = sess->bytes_done;
        gf_dm_data_received(sess, obuffer, 0, GF_FALSE, NULL, NULL);
      }

      if (sess->status==GF_NETIO_DATA_EXCHANGE)
        sess->status = GF_NETIO_DATA_TRANSFERED;
    }
    return;
  }

  gf_assert(sess->hmux_buf.offset<=sess->hmux_buf.size);

  nb_b_pck = sess->hmux_buf.size - sess->hmux_buf.offset;
  if (nb_b_pck > size)
    copy = size;
  else
    copy = nb_b_pck;

  data = sess->hmux_buf.data + sess->hmux_buf.offset;
  memcpy(obuffer, data, copy);
  *nb_bytes = copy;
  //signal we received data (for event triggering / user callbacks)
  gf_dm_data_received(sess, (u8 *) data, copy, GF_FALSE, NULL, NULL);

  if (copy < nb_b_pck) {
    sess->hmux_buf.offset += copy;
  } else {
    sess->hmux_buf.size = sess->hmux_buf.offset = 0;
  }
  gf_assert(sess->hmux_buf.offset<=sess->hmux_buf.size);
}

GF_Err hmux_send_payload(GF_DownloadSession *sess, u8 *data, u32 size)
{
  GF_Err e = GF_OK;
  if (sess->hmux_send_data) {
    e = gf_sk_probe(sess->sock);
    if (e) return e;
    return GF_SERVICE_ERROR;
  }

  if (sess->hmux_stream_id<0)
    return GF_URL_REMOVED;

  gf_mx_p(sess->mx);

  GF_LOG(GF_LOG_DEBUG, GF_LOG_HTTP, ("[%s] Sending %d bytes on stream_id "LLD"\n", sess->log_name, size, sess->hmux_stream_id));

  assert(sess->hmux_send_data==NULL);
  sess->hmux_send_data = data;
  sess->hmux_send_data_len = size;
  if (sess->hmux_data_paused) {
    e = sess->hmux_sess->resume(sess);
    if (e) {
      gf_mx_v(sess->mx);
      return e;
    }
  }
  //if no data, signal end of stream, otherwise regular send
  if (!data || !size) {
    if (sess->local_buf_len) {
      gf_mx_v(sess->mx);
      if (sess->hmux_is_eos) {
        return GF_IP_NETWORK_EMPTY;
      }
      sess->hmux_is_eos = 1;
      sess->hmux_sess->write(sess);
      return GF_OK;
    }
    sess->hmux_is_eos = 1;
    sess->hmux_sess->write(sess);
    //stream_id is not yet 0 in case of PUT/PUSH, stream is closed once we get reply from server
  } else {
    sess->hmux_is_eos = 0;
    //send the data
    e = sess->hmux_sess->send_pending_data(sess);
    if (e && (e!=GF_IP_NETWORK_EMPTY)) {
      gf_mx_v(sess->mx);
      return e;
    }
  }
  gf_mx_v(sess->mx);

  if (!data || !size) {
    if (sess->put_state) {
      //we are done sending
      if (!sess->local_buf_len) {
        sess->put_state = 2;
        sess->status = GF_NETIO_WAIT_FOR_REPLY;
      }
      return GF_OK;
    }
  }
  if ((sess->hmux_stream_id<0) && sess->hmux_send_data)
    return GF_URL_REMOVED;
  return GF_OK;
}

#endif //GPAC_HTTPMUX



void gf_dm_sess_close_hmux(GF_DownloadSession *sess)
{
#ifdef GPAC_HTTPMUX
  if (sess->hmux_sess)
    sess->hmux_sess->close_session(sess);
#endif
}

GF_DownloadSession *gf_dm_sess_new_subsession(GF_DownloadSession *sess, s64 stream_id, void *usr_cbk, GF_Err *e)
{
#ifdef GPAC_HTTPMUX
  GF_DownloadSession *sub_sess;
  if (!sess->hmux_sess) return NULL;
  gf_mx_p(sess->mx);
  sub_sess = gf_dm_sess_new_internal(NULL, NULL, 0, sess->user_proc, usr_cbk, sess->sock, GF_TRUE, e);
  if (!sub_sess) {
    gf_mx_v(sess->mx);
    return NULL;
  }
  gf_list_add(sess->hmux_sess->sessions, sub_sess);
#ifdef GPAC_HAS_SSL
  sub_sess->ssl = sess->ssl;
#endif
  sub_sess->hmux_sess = sess->hmux_sess;
  if (sub_sess->mx) gf_mx_del(sub_sess->mx);
  sub_sess->mx = sess->hmux_sess->mx;
  sub_sess->hmux_stream_id = stream_id;
  sub_sess->status = GF_NETIO_CONNECTED;

  sub_sess->hmux_sess->setup_session(sub_sess, GF_FALSE);

  sub_sess->flags = sess->flags;
  gf_mx_v(sess->mx);
  return sub_sess;
#else
  return NULL;
#endif //GPAC_HTTPMUX
}



#endif //GPAC_DISABLE_NETWORK


Coverage Report

Created: 2025-08-28 06:49

Line	Count	Source (jump to first uncovered line)
1		/*
2		* GPAC Multimedia Framework
3		*
4		* Authors: Jean Le Feuvre
5		* Copyright (c) Telecom ParisTech 2005-2025
6		* All rights reserved
7		*
8		* This file is part of GPAC / downloader sub-project
9		*
10		* GPAC is free software; you can redistribute it and/or modify
11		* it under the terms of the GNU Lesser General Public License as published by
12		* the Free Software Foundation; either version 2, or (at your option)
13		* any later version.
14		*
15		* GPAC is distributed in the hope that it will be useful,
16		* but WITHOUT ANY WARRANTY; without even the implied warranty of
17		* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18		* GNU Lesser General Public License for more details.
19		*
20		* You should have received a copy of the GNU Lesser General Public
21		* License along with this library; see the file COPYING. If not, write to
22		* the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
23		*
24		*/
25
26		#include "downloader.h"
27
28		#ifndef GPAC_DISABLE_NETWORK
29
30		#ifdef GPAC_HTTPMUX
31
32		//detach session from parent h2 / h3 session - the session mutex SHALL be grabbed before calling this
33		void hmux_detach_session(GF_HMUX_Session hmux_sess, GF_DownloadSession sess)
34		{
35		if (!hmux_sess \|\| !sess) return;
36		gf_assert(sess->hmux_sess == hmux_sess);
37		gf_assert(sess->mx);
38
39		gf_list_del_item(hmux_sess->sessions, sess);
40		if (!gf_list_count(hmux_sess->sessions)) {
41		if (hmux_sess->close)
42		hmux_sess->close(hmux_sess);
43
44		#ifdef GPAC_HAS_SSL
45		if (sess->ssl) {
46		GF_LOG(GF_LOG_DEBUG, GF_LOG_HTTP, ("[Downloader] shut down SSL context\n"));
47		SSL_shutdown(sess->ssl);
48		SSL_free(sess->ssl);
49		sess->ssl = NULL;
50		}
51		#endif
52		dm_sess_sk_del(sess);
53		if (hmux_sess->destroy)
54		hmux_sess->destroy(hmux_sess);
55
56		gf_list_del(hmux_sess->sessions);
57		gf_mx_v(hmux_sess->mx);
58		gf_mx_del(sess->mx);
59		gf_free(hmux_sess);
60		} else {
61		GF_DownloadSession *asess = gf_list_get(hmux_sess->sessions, 0);
62		gf_assert(asess->hmux_sess == hmux_sess);
63
64		hmux_sess->net_sess = asess;
65		//swap async buf if any to new active session
66		if (sess->async_buf_alloc) {
67		gf_assert(!asess->async_buf);
68		asess->async_buf = sess->async_buf;
69		asess->async_buf_alloc = sess->async_buf_alloc;
70		asess->async_buf_size = sess->async_buf_size;
71		sess->async_buf_alloc = sess->async_buf_size = 0;
72		sess->async_buf = NULL;
73		}
74		sess->sock = NULL;
75		#ifdef GPAC_HAS_SSL
76		sess->ssl = NULL;
77		#endif
78		gf_mx_v(sess->mx);
79		}
80
81		if (sess->hmux_buf.data) {
82		gf_free(sess->hmux_buf.data);
83		memset(&sess->hmux_buf, 0, sizeof(hmux_reagg_buffer));
84		}
85		sess->hmux_sess = NULL;
86		sess->mx = NULL;
87		}
88
89
90		GF_DownloadSession hmux_get_session(void user_data, s64 stream_id, Bool can_reassign)
91		{
92		u32 i, nb_sess;
93		GF_DownloadSession *first_not_assigned = NULL;
94		GF_HMUX_Session hmux_sess = (GF_HMUX_Session )user_data;
95
96		nb_sess = gf_list_count(hmux_sess->sessions);
97		for (i=0;i<nb_sess; i++) {
98		GF_DownloadSession *s = gf_list_get(hmux_sess->sessions, i);
99		if (s->hmux_stream_id == stream_id)
100		return s;
101
102		if (s->server_mode && (s->hmux_stream_id<0) && !first_not_assigned) {
103		first_not_assigned = s;
104		}
105		}
106		if (can_reassign && first_not_assigned) {
107		first_not_assigned->hmux_stream_id = stream_id;
108		GF_LOG(GF_LOG_DEBUG, GF_LOG_HTTP, ("[%s] reassigning old server session to new stream %d\n", hmux_sess->net_sess->log_name, stream_id));
109		first_not_assigned->status = GF_NETIO_CONNECTED;
110		first_not_assigned->total_size = first_not_assigned->bytes_done = 0;
111		first_not_assigned->hmux_data_done = 0;
112		if (first_not_assigned->remote_path) gf_free(first_not_assigned->remote_path);
113		first_not_assigned->remote_path = NULL;
114		//reset internal reaggregation buffer
115		first_not_assigned->hmux_buf.size = 0;
116
117		gf_dm_sess_clear_headers(first_not_assigned);
118		return first_not_assigned;
119		}
120		return NULL;
121		}
122
123		void hmux_flush_internal_data(GF_DownloadSession *sess, Bool store_in_init)
124		{
125		gf_dm_data_received(sess, (u8 *) sess->hmux_buf.data, sess->hmux_buf.size, store_in_init, NULL, NULL);
126		sess->hmux_buf.size = 0;
127		}
128
129		void hmux_fetch_data(GF_DownloadSession sess, u8 obuffer, u32 size, u32 *nb_bytes)
130		{
131		u32 copy, nb_b_pck;
132		u8 *data;
133		*nb_bytes = 0;
134		if (!sess->hmux_buf.size) {
135		if (sess->hmux_is_eos) {
136		if (!sess->total_size) {
137		sess->total_size = sess->bytes_done;
138		gf_dm_data_received(sess, obuffer, 0, GF_FALSE, NULL, NULL);
139		}
140
141		if (sess->status==GF_NETIO_DATA_EXCHANGE)
142		sess->status = GF_NETIO_DATA_TRANSFERED;
143		}
144		return;
145		}
146
147		gf_assert(sess->hmux_buf.offset<=sess->hmux_buf.size);
148
149		nb_b_pck = sess->hmux_buf.size - sess->hmux_buf.offset;
150		if (nb_b_pck > size)
151		copy = size;
152		else
153		copy = nb_b_pck;
154
155		data = sess->hmux_buf.data + sess->hmux_buf.offset;
156		memcpy(obuffer, data, copy);
157		*nb_bytes = copy;
158		//signal we received data (for event triggering / user callbacks)
159		gf_dm_data_received(sess, (u8 *) data, copy, GF_FALSE, NULL, NULL);
160
161		if (copy < nb_b_pck) {
162		sess->hmux_buf.offset += copy;
163		} else {
164		sess->hmux_buf.size = sess->hmux_buf.offset = 0;
165		}
166		gf_assert(sess->hmux_buf.offset<=sess->hmux_buf.size);
167		}
168
169		GF_Err hmux_send_payload(GF_DownloadSession sess, u8 data, u32 size)
170		{
171		GF_Err e = GF_OK;
172		if (sess->hmux_send_data) {
173		e = gf_sk_probe(sess->sock);
174		if (e) return e;
175		return GF_SERVICE_ERROR;
176		}
177
178		if (sess->hmux_stream_id<0)
179		return GF_URL_REMOVED;
180
181		gf_mx_p(sess->mx);
182
183		GF_LOG(GF_LOG_DEBUG, GF_LOG_HTTP, ("[%s] Sending %d bytes on stream_id "LLD"\n", sess->log_name, size, sess->hmux_stream_id));
184
185		assert(sess->hmux_send_data==NULL);
186		sess->hmux_send_data = data;
187		sess->hmux_send_data_len = size;
188		if (sess->hmux_data_paused) {
189		e = sess->hmux_sess->resume(sess);
190		if (e) {
191		gf_mx_v(sess->mx);
192		return e;
193		}
194		}
195		//if no data, signal end of stream, otherwise regular send
196		if (!data \|\| !size) {
197		if (sess->local_buf_len) {
198		gf_mx_v(sess->mx);
199		if (sess->hmux_is_eos) {
200		return GF_IP_NETWORK_EMPTY;
201		}
202		sess->hmux_is_eos = 1;
203		sess->hmux_sess->write(sess);
204		return GF_OK;
205		}
206		sess->hmux_is_eos = 1;
207		sess->hmux_sess->write(sess);
208		//stream_id is not yet 0 in case of PUT/PUSH, stream is closed once we get reply from server
209		} else {
210		sess->hmux_is_eos = 0;
211		//send the data
212		e = sess->hmux_sess->send_pending_data(sess);
213		if (e && (e!=GF_IP_NETWORK_EMPTY)) {
214		gf_mx_v(sess->mx);
215		return e;
216		}
217		}
218		gf_mx_v(sess->mx);
219
220		if (!data \|\| !size) {
221		if (sess->put_state) {
222		//we are done sending
223		if (!sess->local_buf_len) {
224		sess->put_state = 2;
225		sess->status = GF_NETIO_WAIT_FOR_REPLY;
226		}
227		return GF_OK;
228		}
229		}
230		if ((sess->hmux_stream_id<0) && sess->hmux_send_data)
231		return GF_URL_REMOVED;
232		return GF_OK;
233		}
234
235		#endif //GPAC_HTTPMUX
236
237
238
239		void gf_dm_sess_close_hmux(GF_DownloadSession *sess)
240	0	{
241		#ifdef GPAC_HTTPMUX
242		if (sess->hmux_sess)
243		sess->hmux_sess->close_session(sess);
244		#endif
245	0	}
246
247		GF_DownloadSession gf_dm_sess_new_subsession(GF_DownloadSession sess, s64 stream_id, void usr_cbk, GF_Err e)
248	0	{
249		#ifdef GPAC_HTTPMUX
250		GF_DownloadSession *sub_sess;
251		if (!sess->hmux_sess) return NULL;
252		gf_mx_p(sess->mx);
253		sub_sess = gf_dm_sess_new_internal(NULL, NULL, 0, sess->user_proc, usr_cbk, sess->sock, GF_TRUE, e);
254		if (!sub_sess) {
255		gf_mx_v(sess->mx);
256		return NULL;
257		}
258		gf_list_add(sess->hmux_sess->sessions, sub_sess);
259		#ifdef GPAC_HAS_SSL
260		sub_sess->ssl = sess->ssl;
261		#endif
262		sub_sess->hmux_sess = sess->hmux_sess;
263		if (sub_sess->mx) gf_mx_del(sub_sess->mx);
264		sub_sess->mx = sess->hmux_sess->mx;
265		sub_sess->hmux_stream_id = stream_id;
266		sub_sess->status = GF_NETIO_CONNECTED;
267
268		sub_sess->hmux_sess->setup_session(sub_sess, GF_FALSE);
269
270		sub_sess->flags = sess->flags;
271		gf_mx_v(sess->mx);
272		return sub_sess;
273		#else
274	0	return NULL;
275	0	#endif //GPAC_HTTPMUX
276	0	}
277
278
279
280		#endif //GPAC_DISABLE_NETWORK
281