upmpdcli / Code / Diff of /src/mediaserver/cdplugins/streamproxy.cpp

Diff of /src/mediaserver/cdplugins/streamproxy.cpp [3c97d1] .. [c44069]

Switch to unified view


...
 *   59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
 */

#include "streamproxy.h"
#include "netfetch.h"

#include "log.h"
#include "smallut.h"
#include "chrono.h"

#include <fcntl.h>
...

using namespace std;

class ContentReader {
public:
    ContentReader(std::unique_ptr<NetFetch> ftchr, int cfd)
        : fetcher(std::move(ftchr)), connfd(cfd) {
        
    }

    ~ContentReader() {
        LOGDEB1("ContentReader::~ContentReader\n");
        // This should not be necessary but see comments in
        // tstcurlfetch code
        fetcher = std::unique_ptr<NetFetch>();
...
    int connfd{-1};
    int killafterms{-1};
    Chrono chron;
};






ssize_t ContentReader::contentRead(uint64_t pos, char *obuf, size_t max)
{
    LOGDEB1("ContentReader::contentRead: pos "<<pos << " max " << max << endl);
    if (normalEOS) {
...

        // Compute destination url
        unordered_map<string,string>querydata;
        MHD_get_connection_values(mhdconn, MHD_GET_ARGUMENT_KIND,
                                  &mapvalues_cb, &querydata);

        // Request the method (redirect or proxy), and the fetcher if
        // we are proxying.
        string url(_url);
        std::unique_ptr<NetFetch> fetcher;
        UrlTransReturn ret = urltrans(url, querydata, fetcher);
        


        if (ret == Error) {
            return MHD_NO;
        } else if (ret == Redirect) {
            struct MHD_Response *response =
                MHD_create_response_from_buffer(0, 0, MHD_RESPMEM_PERSISTENT);
...
            int ret = MHD_queue_response(mhdconn, 302, response);
            MHD_destroy_response(response);
            return ret;
        }

        // The connection fd thing is strictly for debug/diag: faking
        // a connection loss so that the client side can exercise the
        // retry code.
        int cfd{-1};
        const union MHD_ConnectionInfo *cinf =
            MHD_get_connection_info(mhdconn, MHD_CONNECTION_INFO_CONNECTION_FD);
        if (nullptr == cinf) {
            LOGERR("StreamProxy::answerConn: can't get connection fd\n");
        } else {
            cfd = cinf->connect_fd;
        }

        // Create/Start the fetch transaction, and wait for the headers.
        LOGDEB0("StreamProxy::answerConn: starting fetch for " << url << endl);
        auto reader = new ContentReader(std::move(fetcher), cfd);
        if (killafterms > 0) {
            reader->killafterms = killafterms;
            killafterms = -1;
        }
        reader->fetcher->start(&reader->queue, offset);
        *con_cls = reader;

        LOGDEB1("StreamProxy::answerConn: returning after 1st call\n");
        return MHD_YES;
        // End first call
    }


	a/src/mediaserver/cdplugins/streamproxy.cpp		b/src/mediaserver/cdplugins/streamproxy.cpp
	...		...
15	* 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.	15	* 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
16	*/	16	*/
17		17
18	#include "streamproxy.h"	18	#include "streamproxy.h"
19	#include "netfetch.h"	19	#include "netfetch.h"
20	#include "curlfetch.h"
21	#include "log.h"	20	#include "log.h"
22	#include "smallut.h"	21	#include "smallut.h"
23	#include "chrono.h"	22	#include "chrono.h"
24		23
25	#include <fcntl.h>	24	#include <fcntl.h>
	...		...
31		30
32	using namespace std;	31	using namespace std;
33		32
34	class ContentReader {	33	class ContentReader {
35	public:	34	public:
36	ContentReader(NetFetch *fetcher, int cfd);	35	ContentReader(std::unique_ptr<NetFetch> ftchr, int cfd)
		36	: fetcher(std::move(ftchr)), connfd(cfd) {
		37
		38	}
		39
37	~ContentReader() {	40	~ContentReader() {
38	LOGDEB1("ContentReader::~ContentReader\n");	41	LOGDEB1("ContentReader::~ContentReader\n");
39	// This should not be necessary but see comments in	42	// This should not be necessary but see comments in
40	// tstcurlfetch code	43	// tstcurlfetch code
41	fetcher = std::unique_ptr<NetFetch>();	44	fetcher = std::unique_ptr<NetFetch>();
	...		...
49	int connfd{-1};	52	int connfd{-1};
50	int killafterms{-1};	53	int killafterms{-1};
51	Chrono chron;	54	Chrono chron;
52	};	55	};
53		56
54	ContentReader::ContentReader(NetFetch *f, int cfd)
55	: fetcher(std::unique_ptr<NetFetch>(f)), connfd(cfd)
56	{
57	}
58		57
59	ssize_t ContentReader::contentRead(uint64_t pos, char *obuf, size_t max)	58	ssize_t ContentReader::contentRead(uint64_t pos, char *obuf, size_t max)
60	{	59	{
61	LOGDEB1("ContentReader::contentRead: pos "<<pos << " max " << max << endl);	60	LOGDEB1("ContentReader::contentRead: pos "<<pos << " max " << max << endl);
62	if (normalEOS) {	61	if (normalEOS) {
	...		...
309		308
310	// Compute destination url	309	// Compute destination url
311	unordered_map<string,string>querydata;	310	unordered_map<string,string>querydata;
312	MHD_get_connection_values(mhdconn, MHD_GET_ARGUMENT_KIND,	311	MHD_get_connection_values(mhdconn, MHD_GET_ARGUMENT_KIND,
313	&mapvalues_cb, &querydata);	312	&mapvalues_cb, &querydata);
		313
		314	// Request the method (redirect or proxy), and the fetcher if
		315	// we are proxying.
		316	string url(_url);
		317	std::unique_ptr<NetFetch> fetcher;
		318	UrlTransReturn ret = urltrans(url, querydata, fetcher);
314		319
315	string url(_url);
316	UrlTransReturn ret = urltrans(url, querydata);
317	if (ret == Error) {	320	if (ret == Error) {
318	return MHD_NO;	321	return MHD_NO;
319	} else if (ret == Redirect) {	322	} else if (ret == Redirect) {
320	struct MHD_Response *response =	323	struct MHD_Response *response =
321	MHD_create_response_from_buffer(0, 0, MHD_RESPMEM_PERSISTENT);	324	MHD_create_response_from_buffer(0, 0, MHD_RESPMEM_PERSISTENT);
	...		...
327	int ret = MHD_queue_response(mhdconn, 302, response);	330	int ret = MHD_queue_response(mhdconn, 302, response);
328	MHD_destroy_response(response);	331	MHD_destroy_response(response);
329	return ret;	332	return ret;
330	}	333	}
331		334
332	// Create/Start the fetch transaction, and wait for the headers.	335	// The connection fd thing is strictly for debug/diag: faking
333	LOGDEB0("StreamProxy::answerConn: starting fetch for " << url << endl);	336	// a connection loss so that the client side can exercise the
		337	// retry code.
334	int cfd{-1};	338	int cfd{-1};
335	const union MHD_ConnectionInfo *cinf =	339	const union MHD_ConnectionInfo *cinf =
336	MHD_get_connection_info(mhdconn, MHD_CONNECTION_INFO_CONNECTION_FD);	340	MHD_get_connection_info(mhdconn, MHD_CONNECTION_INFO_CONNECTION_FD);
337	if (nullptr == cinf) {	341	if (nullptr == cinf) {
338	LOGERR("StreamProxy::answerConn: can't get connection fd\n");	342	LOGERR("StreamProxy::answerConn: can't get connection fd\n");
339	} else {	343	} else {
340	cfd = cinf->connect_fd;	344	cfd = cinf->connect_fd;
341	}	345	}
		346
		347	// Create/Start the fetch transaction, and wait for the headers.
		348	LOGDEB0("StreamProxy::answerConn: starting fetch for " << url << endl);
342	auto reader = new ContentReader(new CurlFetch(url), cfd);	349	auto reader = new ContentReader(std::move(fetcher), cfd);
343	if (killafterms > 0) {	350	if (killafterms > 0) {
344	reader->killafterms = killafterms;	351	reader->killafterms = killafterms;
345	killafterms = -1;	352	killafterms = -1;
346	}	353	}
347	reader->fetcher->start(&reader->queue, offset);	354	reader->fetcher->start(&reader->queue, offset);
348	*con_cls = reader;	355	*con_cls = reader;
		356
349	LOGDEB1("StreamProxy::answerConn: returning after 1st call\n");	357	LOGDEB1("StreamProxy::answerConn: returning after 1st call\n");
350	return MHD_YES;	358	return MHD_YES;
351	// End first call	359	// End first call
352	}	360	}
353		361