1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
#ifndef _HTTPINPUT_H
#define _HTTPINPUT_H 1
#include <stddef.h>
#include <sys/types.h>
#include <mutex>
#include <string>
#include <vector>
#include "input.h"
#include "metacube2.h"
#include "stream.h"
class InputProto;
// Despite the name, covers input over both HTTP and pipes, both typically
// wrapped in Metacube.
class HTTPInput : public Input {
public:
HTTPInput(const std::string &url, Input::Encoding encoding);
// Serialization/deserialization.
HTTPInput(const InputProto &serialized);
virtual InputProto serialize() const;
virtual void close_socket();
virtual std::string get_url() const { return url; }
virtual void add_destination(int stream_index);
virtual InputStats get_stats() const;
private:
// Actually does the download.
virtual void do_work();
// Open a socket that connects to the given host and port. Does DNS resolving.
int lookup_and_connect(const std::string &host, const std::string &port);
// Open a child process with the given command line (given to /bin/sh).
// Returns a pipe to its standard output.
int open_child_process(const std::string &cmdline);
// Parses a HTTP response. Returns false if it not a 200.
bool parse_response(const std::string &response);
// Stores the given data, looks for Metacube blocks (skipping data if needed),
// and calls process_block() for each one.
void process_data(char *ptr, size_t bytes);
// Drops <num_bytes> bytes from the head of <pending_data>,
// and outputs a warning.
void drop_pending_data(size_t num_bytes);
void process_metacube_metadata_block(const metacube2_block_header &hdr, const char *payload, uint32_t payload_size);
enum State {
NOT_CONNECTED,
SENDING_REQUEST,
RECEIVING_HEADER,
RECEIVING_DATA,
CLOSING_SOCKET, // Due to error.
};
State state;
std::vector<int> stream_indices;
// The URL and its parsed components.
std::string url;
std::string host, port, path;
// What the input stream is to be interpreted as (normally Metacube).
Input::Encoding encoding;
// The HTTP request, with headers and all.
// Only relevant for SENDING_REQUEST.
std::string request;
// How many bytes we've sent of the request so far.
// Only relevant for SENDING_REQUEST.
size_t request_bytes_sent;
// The HTTP response we've received so far. Only relevant for RECEIVING_HEADER.
std::string response;
// The HTTP response headers we want to give clients for this input.
std::string http_header;
// The stream heder we want to give clients for this input.
std::string stream_header;
// Data we have received but not fully processed yet.
std::vector<char> pending_data;
// If <pending_data> starts with a Metacube header,
// this is true.
bool has_metacube_header = false;
// The socket we are downloading on (or -1).
int sock = -1;
// pid of the child process (or -1).
pid_t child_pid = -1;
// Mutex protecting <stats>.
mutable std::mutex stats_mutex;
// The current statistics for this connection. Protected by <stats_mutex>.
InputStats stats;
// Number of (started) connection attempts since last data byte was successfully read.
unsigned num_connection_attempts = 0;
// If set, don't log anything related to connections.
// (Only set if we've had enough unsuccessful connection attempts.)
bool suppress_logging = false;
// Last time we made a connection with logging enabled.
// (Initially at some point before the epoch.)
timespec last_verbose_connection { -3600, 0 };
// If we've received a Metacube2 PTS metadata block, it belongs to the
// next regular block we receive, and is stored here in the meantime.
// If we haven't received one yet (or we've already received the
// corresponding data block), this is empty, ie., timebase_num == 0.
RationalPTS next_block_pts;
};
#endif // !defined(_HTTPINPUT_H)
|
