use tcp client for raw connect.

pull/510/head
winlin 9 years ago
parent 20fcfb3eee
commit a9bb6061c3

@ -119,9 +119,8 @@ SrsDynamicHttpConn::SrsDynamicHttpConn(IConnectionManager* cm, st_netfd_t fd, Sr
{ {
req = NULL; req = NULL;
io = NULL; transport = new SrsTcpClient();
client = NULL; client = NULL;
stfd = NULL;
stream_id = 0; stream_id = 0;
pprint = SrsPithyPrint::create_caster(); pprint = SrsPithyPrint::create_caster();
@ -131,6 +130,7 @@ SrsDynamicHttpConn::~SrsDynamicHttpConn()
{ {
close(); close();
srs_freep(transport);
srs_freep(pprint); srs_freep(pprint);
} }
@ -261,7 +261,7 @@ int SrsDynamicHttpConn::connect()
// when ok, ignore. // when ok, ignore.
// TODO: FIXME: should reconnect when disconnected. // TODO: FIXME: should reconnect when disconnected.
if (io || client) { if (transport->connected()) {
return ret; return ret;
} }
@ -273,12 +273,10 @@ int SrsDynamicHttpConn::connect()
} }
// connect host. // connect host.
if ((ret = srs_socket_connect(req->host, req->port, ST_UTIME_NO_TIMEOUT, &stfd)) != ERROR_SUCCESS) { if ((ret = transport->connect(req->host, req->port, ST_UTIME_NO_TIMEOUT)) != ERROR_SUCCESS) {
srs_error("mpegts: connect server %s:%d failed. ret=%d", req->host.c_str(), req->port, ret);
return ret; return ret;
} }
io = new SrsStSocket(stfd); client = new SrsRtmpClient(transport);
client = new SrsRtmpClient(io);
client->set_recv_timeout(SRS_CONSTS_RTMP_RECV_TIMEOUT_US); client->set_recv_timeout(SRS_CONSTS_RTMP_RECV_TIMEOUT_US);
client->set_send_timeout(SRS_CONSTS_RTMP_SEND_TIMEOUT_US); client->set_send_timeout(SRS_CONSTS_RTMP_SEND_TIMEOUT_US);
@ -360,10 +358,10 @@ int SrsDynamicHttpConn::connect_app(string ep_server, int ep_port)
void SrsDynamicHttpConn::close() void SrsDynamicHttpConn::close()
{ {
transport->close();
srs_freep(client); srs_freep(client);
srs_freep(io);
srs_freep(req); srs_freep(req);
srs_close_stfd(stfd);
} }
SrsHttpFileReader::SrsHttpFileReader(ISrsHttpResponseReader* h) SrsHttpFileReader::SrsHttpFileReader(ISrsHttpResponseReader* h)

@ -43,6 +43,7 @@ class SrsRequest;
class SrsPithyPrint; class SrsPithyPrint;
class ISrsHttpResponseReader; class ISrsHttpResponseReader;
class SrsFlvDecoder; class SrsFlvDecoder;
class SrsTcpClient;
#include <srs_app_st.hpp> #include <srs_app_st.hpp>
#include <srs_app_listener.hpp> #include <srs_app_listener.hpp>
@ -86,8 +87,7 @@ private:
SrsPithyPrint* pprint; SrsPithyPrint* pprint;
private: private:
SrsRequest* req; SrsRequest* req;
st_netfd_t stfd; SrsTcpClient* transport;
SrsStSocket* io;
SrsRtmpClient* client; SrsRtmpClient* client;
int stream_id; int stream_id;
public: public:

@ -28,6 +28,7 @@ using namespace std;
#include <srs_kernel_error.hpp> #include <srs_kernel_error.hpp>
#include <srs_kernel_log.hpp> #include <srs_kernel_log.hpp>
#include <srs_app_utility.hpp>
namespace internal namespace internal
{ {
@ -411,18 +412,106 @@ int SrsStSocket::writev(const iovec *iov, int iov_size, ssize_t* nwrite)
SrsTcpClient::SrsTcpClient() SrsTcpClient::SrsTcpClient()
{ {
io = NULL;
stfd = NULL;
} }
SrsTcpClient::~SrsTcpClient() SrsTcpClient::~SrsTcpClient()
{ {
close();
}
bool SrsTcpClient::connected()
{
return io;
} }
int SrsTcpClient::connect(string host, int port, int64_t timeout) int SrsTcpClient::connect(string host, int port, int64_t timeout)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
// when connected, ignore.
if (io) {
return ret;
}
// connect host.
if ((ret = srs_socket_connect(host, port, timeout, &stfd)) != ERROR_SUCCESS) {
srs_error("mpegts: connect server %s:%d failed. ret=%d", host.c_str(), port, ret);
return ret;
}
io = new SrsStSocket(stfd);
return ret; return ret;
} }
void SrsTcpClient::close()
{
// when closed, ignore.
if (!io) {
return;
}
srs_freep(io);
srs_close_stfd(stfd);
}
bool SrsTcpClient::is_never_timeout(int64_t timeout_us)
{
return io->is_never_timeout(timeout_us);
}
void SrsTcpClient::set_recv_timeout(int64_t timeout_us)
{
io->set_recv_timeout(timeout_us);
}
int64_t SrsTcpClient::get_recv_timeout()
{
return io->get_recv_timeout();
}
void SrsTcpClient::set_send_timeout(int64_t timeout_us)
{
io->set_send_timeout(timeout_us);
}
int64_t SrsTcpClient::get_send_timeout()
{
return io->get_send_timeout();
}
int64_t SrsTcpClient::get_recv_bytes()
{
return io->get_recv_bytes();
}
int64_t SrsTcpClient::get_send_bytes()
{
return io->get_send_bytes();
}
int SrsTcpClient::read(void* buf, size_t size, ssize_t* nread)
{
return io->read(buf, size, nread);
}
int SrsTcpClient::read_fully(void* buf, size_t size, ssize_t* nread)
{
return io->read_fully(buf, size, nread);
}
int SrsTcpClient::write(void* buf, size_t size, ssize_t* nwrite)
{
return io->write(buf, size, nwrite);
}
int SrsTcpClient::writev(const iovec *iov, int iov_size, ssize_t* nwrite)
{
return io->writev(iov, iov_size, nwrite);
}
#ifdef __linux__ #ifdef __linux__
#include <sys/epoll.h> #include <sys/epoll.h>

@ -208,19 +208,46 @@ public:
* the common tcp client, to connect to specified TCP server, * the common tcp client, to connect to specified TCP server,
* reconnect and close the connection. * reconnect and close the connection.
*/ */
class SrsTcpClient class SrsTcpClient : public ISrsProtocolReaderWriter
{ {
private:
st_netfd_t stfd;
SrsStSocket* io;
public: public:
SrsTcpClient(); SrsTcpClient();
virtual ~SrsTcpClient(); virtual ~SrsTcpClient();
public:
/**
* whether connected to server.
*/
virtual bool connected();
public: public:
/** /**
* connect to server over TCP. * connect to server over TCP.
* @param host the ip or hostname of server. * @param host the ip or hostname of server.
* @param port the port to connect to. * @param port the port to connect to.
* @param timeout the timeout in us. * @param timeout the timeout in us.
* @remark ignore when connected.
*/ */
virtual int connect(std::string host, int port, int64_t timeout); virtual int connect(std::string host, int port, int64_t timeout);
/**
* close the connection.
* @remark ignore when closed.
*/
virtual void close();
// interface ISrsProtocolReaderWriter
public:
virtual bool is_never_timeout(int64_t timeout_us);
virtual void set_recv_timeout(int64_t timeout_us);
virtual int64_t get_recv_timeout();
virtual void set_send_timeout(int64_t timeout_us);
virtual int64_t get_send_timeout();
virtual int64_t get_recv_bytes();
virtual int64_t get_send_bytes();
virtual int read(void* buf, size_t size, ssize_t* nread);
virtual int read_fully(void* buf, size_t size, ssize_t* nread);
virtual int write(void* buf, size_t size, ssize_t* nwrite);
virtual int writev(const iovec *iov, int iov_size, ssize_t* nwrite);
}; };
// initialize st, requires epoll. // initialize st, requires epoll.

Loading…
Cancel
Save