Commit 6c408d5c authored by ciody's avatar ciody

Merge remote-tracking branch 'upstream/master'

parents 8dbc68bb 2ab2b548
...@@ -30,8 +30,6 @@ namespace Http { ...@@ -30,8 +30,6 @@ namespace Http {
class ConnectionPool; class ConnectionPool;
class Transport; class Transport;
std::pair<StringView, StringView> splitUrl(const std::string& url);
struct Connection : public std::enable_shared_from_this<Connection> { struct Connection : public std::enable_shared_from_this<Connection> {
friend class ConnectionPool; friend class ConnectionPool;
...@@ -96,8 +94,7 @@ struct Connection : public std::enable_shared_from_this<Connection> { ...@@ -96,8 +94,7 @@ struct Connection : public std::enable_shared_from_this<Connection> {
Async::Rejection reject, Async::Rejection reject,
OnDone onDone); OnDone onDone);
Fd fd; Fd fd() const;
void handleResponsePacket(const char* buffer, size_t totalBytes); void handleResponsePacket(const char* buffer, size_t totalBytes);
void handleError(const char* error); void handleError(const char* error);
void handleTimeout(); void handleTimeout();
...@@ -124,6 +121,8 @@ private: ...@@ -124,6 +121,8 @@ private:
OnDone onDone; OnDone onDone;
}; };
Fd fd_;
struct sockaddr_in saddr; struct sockaddr_in saddr;
std::unique_ptr<RequestEntry> requestEntry; std::unique_ptr<RequestEntry> requestEntry;
std::atomic<uint32_t> state_; std::atomic<uint32_t> state_;
......
...@@ -12,8 +12,7 @@ ...@@ -12,8 +12,7 @@
#include <limits> #include <limits>
#include <sys/socket.h> #include <sys/socket.h>
#include <netdb.h>
#include <pistache/common.h>
#ifndef _KERNEL_FASTOPEN #ifndef _KERNEL_FASTOPEN
#define _KERNEL_FASTOPEN #define _KERNEL_FASTOPEN
...@@ -65,6 +64,7 @@ private: ...@@ -65,6 +64,7 @@ private:
class Port { class Port {
public: public:
Port(uint16_t port = 0); Port(uint16_t port = 0);
explicit Port(const std::string& data);
operator uint16_t() const { return port; } operator uint16_t() const { return port; }
...@@ -123,6 +123,18 @@ private: ...@@ -123,6 +123,18 @@ private:
uint16_t h; uint16_t h;
}; };
class AddressParser {
public:
explicit AddressParser(const std::string& data);
const std::string& rawHost() const;
const std::string& rawPort() const;
int family() const;
private:
std::string host_;
std::string port_;
int family_;
};
class Address { class Address {
public: public:
Address(); Address();
......
...@@ -40,8 +40,8 @@ public: ...@@ -40,8 +40,8 @@ public:
Peer(const Address& addr); Peer(const Address& addr);
~Peer() {} ~Peer() {}
Address address() const; const Address& address() const;
std::string hostname() const; const std::string& hostname() const;
void associateFd(Fd fd); void associateFd(Fd fd);
Fd fd() const; Fd fd() const;
......
...@@ -6,6 +6,8 @@ ...@@ -6,6 +6,8 @@
#pragma once #pragma once
#include <pistache/os.h>
#include <cstddef> #include <cstddef>
#include <stdexcept> #include <stdexcept>
#include <cstring> #include <cstring>
...@@ -15,7 +17,6 @@ ...@@ -15,7 +17,6 @@
#include <iostream> #include <iostream>
#include <string> #include <string>
#include <pistache/os.h>
namespace Pistache { namespace Pistache {
...@@ -60,7 +61,6 @@ public: ...@@ -60,7 +61,6 @@ public:
const CharT* gptr = this->gptr(); const CharT* gptr = this->gptr();
return *(gptr + 1); return *(gptr + 1);
} }
}; };
template<typename CharT = char> template<typename CharT = char>
...@@ -74,7 +74,6 @@ public: ...@@ -74,7 +74,6 @@ public:
RawStreamBuf(char* begin, size_t len) { RawStreamBuf(char* begin, size_t len) {
Base::setg(begin, begin, begin + len); Base::setg(begin, begin, begin + len);
} }
}; };
// Make the buffer dynamic // Make the buffer dynamic
...@@ -123,33 +122,17 @@ private: ...@@ -123,33 +122,17 @@ private:
template<typename CharT> template<typename CharT>
size_t ArrayStreamBuf<CharT>::maxSize = Const::DefaultMaxPayload; size_t ArrayStreamBuf<CharT>::maxSize = Const::DefaultMaxPayload;
struct Buffer { struct Buffer
Buffer() {
: data(nullptr) Buffer();
, len(0) Buffer(std::string data, int length, bool isDetached = false);
, isOwned(false) Buffer(const char * data, int length, bool isDetached = false);
{ }
Buffer(const char * const _data, size_t _len, bool _own = false) Buffer detach(size_t fromIndex);
: data(_data)
, len(_len)
, isOwned(_own)
{ }
Buffer detach(size_t fromIndex = 0) const { std::string data;
if (fromIndex > len) size_t length;
throw std::invalid_argument("Invalid index (> len)"); int isDetached;
size_t retainedLen = len - fromIndex;
char *newData = new char[retainedLen];
std::copy(data + fromIndex, data + len, newData);
return Buffer(newData, retainedLen, true);
}
const char* const data;
const size_t len;
const bool isOwned;
}; };
struct FileBuffer { struct FileBuffer {
...@@ -165,7 +148,7 @@ struct FileBuffer { ...@@ -165,7 +148,7 @@ struct FileBuffer {
Fd fd() const { return fd_; } Fd fd() const { return fd_; }
size_t size() const { return size_; } size_t size() const { return size_; }
private: private:
std::string fileName_; std::string fileName_;
Fd fd_; Fd fd_;
size_t size_; size_t size_;
...@@ -206,7 +189,7 @@ public: ...@@ -206,7 +189,7 @@ public:
} }
Buffer buffer() const { Buffer buffer() const {
return Buffer(data_.data(), pptr() - &data_[0]); return Buffer((char*) data_.data(), pptr() - &data_[0]);
} }
void clear() { void clear() {
...@@ -214,7 +197,7 @@ public: ...@@ -214,7 +197,7 @@ public:
this->setp(&data_[0], &data_[0] + data_.capacity()); this->setp(&data_[0], &data_[0] + data_.capacity());
} }
protected: protected:
int_type overflow(int_type ch); int_type overflow(int_type ch);
private: private:
...@@ -261,7 +244,7 @@ public: ...@@ -261,7 +244,7 @@ public:
return gptr; return gptr;
} }
private: private:
StreamCursor& cursor; StreamCursor& cursor;
size_t position; size_t position;
char *eback; char *eback;
......
...@@ -80,14 +80,14 @@ private: ...@@ -80,14 +80,14 @@ private:
enum Type { Raw, File }; enum Type { Raw, File };
explicit BufferHolder(const Buffer& buffer, off_t offset = 0) explicit BufferHolder(const Buffer& buffer, off_t offset = 0)
: u(buffer) : _raw(buffer)
, size_(buffer.len) , size_(buffer.length)
, offset_(offset) , offset_(offset)
, type(Raw) , type(Raw)
{ } { }
explicit BufferHolder(const FileBuffer& buffer, off_t offset = 0) explicit BufferHolder(const FileBuffer& buffer, off_t offset = 0)
: u(buffer.fd()) : _fd(buffer.fd())
, size_(buffer.size()) , size_(buffer.size())
, offset_(offset) , offset_(offset)
, type(File) , type(File)
...@@ -101,44 +101,38 @@ private: ...@@ -101,44 +101,38 @@ private:
Fd fd() const { Fd fd() const {
if (!isFile()) if (!isFile())
throw std::runtime_error("Tried to retrieve fd of a non-filebuffer"); throw std::runtime_error("Tried to retrieve fd of a non-filebuffer");
return _fd;
return u.fd;
} }
Buffer raw() const { Buffer raw() const {
if (!isRaw()) if (!isRaw())
throw std::runtime_error("Tried to retrieve raw data of a non-buffer"); throw std::runtime_error("Tried to retrieve raw data of a non-buffer");
return _raw;
return u.raw;
} }
BufferHolder detach(size_t offset = 0) const { BufferHolder detach(size_t offset = 0) {
if (!isRaw()) if (!isRaw())
return BufferHolder(u.fd, size_, offset); return BufferHolder(_fd, size_, offset);
if (u.raw.isOwned) if (_raw.isDetached)
return BufferHolder(u.raw, offset); return BufferHolder(_raw, offset);
auto detached = u.raw.detach(offset); auto detached = _raw.detach(offset);
return BufferHolder(detached); return BufferHolder(detached);
} }
private: private:
BufferHolder(Fd fd, size_t size, off_t offset = 0) BufferHolder(Fd fd, size_t size, off_t offset = 0)
: u(fd) : _fd(fd)
, size_(size) , size_(size)
, offset_(offset) , offset_(offset)
, type(File) , type(File)
{ } { }
union U { Buffer _raw;
Buffer raw; Fd _fd;
Fd fd;
U(Buffer buffer) : raw(buffer) { }
U(Fd fd_) : fd(fd_) { }
} u;
size_t size_= 0; size_t size_= 0;
off_t offset_ = 0; off_t offset_ = 0;
Type type; Type type;
...@@ -225,9 +219,11 @@ private: ...@@ -225,9 +219,11 @@ private:
std::shared_ptr<Peer>& getPeer(Polling::Tag tag); std::shared_ptr<Peer>& getPeer(Polling::Tag tag);
void void
armTimerMs(Fd fd, armTimerMs(
std::chrono::milliseconds value, Fd fd,
Async::Deferred<uint64_t> deferred); std::chrono::milliseconds value,
Async::Deferred<uint64_t> deferred
);
void armTimerMsImpl(TimerEntry entry); void armTimerMsImpl(TimerEntry entry);
......
...@@ -26,22 +26,25 @@ namespace Http { ...@@ -26,22 +26,25 @@ namespace Http {
static constexpr const char* UA = "pistache/0.1"; static constexpr const char* UA = "pistache/0.1";
std::pair<StringView, StringView> namespace
splitUrl(const std::string& url) { {
RawStreamBuf<char> buf(const_cast<char *>(&url[0]), url.size()); std::pair<StringView, StringView> splitUrl(const std::string& url)
StreamCursor cursor(&buf); {
RawStreamBuf<char> buf(const_cast<char *>(&url[0]), url.size());
StreamCursor cursor(&buf);
match_string("http://", std::strlen("http://"), cursor); match_string("http://", std::strlen("http://"), cursor);
match_string("www", std::strlen("www"), cursor); match_string("www", std::strlen("www"), cursor);
match_literal('.', cursor); match_literal('.', cursor);
StreamCursor::Token hostToken(cursor); StreamCursor::Token hostToken(cursor);
match_until({ '?', '/' }, cursor); match_until({ '?', '/' }, cursor);
StringView host(hostToken.rawText(), hostToken.size()); StringView host(hostToken.rawText(), hostToken.size());
StringView page(cursor.offset(), buf.endptr()); StringView page(cursor.offset(), buf.endptr());
return std::make_pair(std::move(host), std::move(page)); return std::make_pair(std::move(host), std::move(page));
}
} }
struct ExceptionPrinter { struct ExceptionPrinter {
...@@ -174,7 +177,7 @@ Transport::onReady(const Aio::FdSet& fds) { ...@@ -174,7 +177,7 @@ Transport::onReady(const Aio::FdSet& fds) {
// We are connected, we can start reading data now // We are connected, we can start reading data now
auto connection = connIt->second.connection.lock(); auto connection = connIt->second.connection.lock();
if (connection) { if (connection) {
reactor()->modifyFd(key(), connection->fd, NotifyOn::Read); reactor()->modifyFd(key(), connection->fd(), NotifyOn::Read);
} else { } else {
throw std::runtime_error("Connection error"); throw std::runtime_error("Connection error");
} }
...@@ -228,7 +231,7 @@ Transport::asyncSendRequestImpl( ...@@ -228,7 +231,7 @@ Transport::asyncSendRequestImpl(
if (!conn) if (!conn)
throw std::runtime_error("Send request error"); throw std::runtime_error("Send request error");
auto fd = conn->fd; auto fd = conn->fd();
ssize_t totalWritten = 0; ssize_t totalWritten = 0;
for (;;) { for (;;) {
...@@ -284,17 +287,17 @@ Transport::handleConnectionQueue() { ...@@ -284,17 +287,17 @@ Transport::handleConnectionQueue() {
if (!conn) { if (!conn) {
throw std::runtime_error("Connection error"); throw std::runtime_error("Connection error");
} }
int res = ::connect(conn->fd, data->getAddr(), data->addr_len); int res = ::connect(conn->fd(), data->getAddr(), data->addr_len);
if (res == -1) { if (res == -1) {
if (errno == EINPROGRESS) { if (errno == EINPROGRESS) {
reactor()->registerFdOneShot(key(), conn->fd, NotifyOn::Write | NotifyOn::Hangup | NotifyOn::Shutdown); reactor()->registerFdOneShot(key(), conn->fd(), NotifyOn::Write | NotifyOn::Hangup | NotifyOn::Shutdown);
} }
else { else {
data->reject(Error::system("Failed to connect")); data->reject(Error::system("Failed to connect"));
continue; continue;
} }
} }
connections.insert(std::make_pair(conn->fd, std::move(*data))); connections.insert(std::make_pair(conn->fd(), std::move(*data)));
} }
} }
...@@ -305,7 +308,7 @@ Transport::handleIncoming(std::shared_ptr<Connection> connection) { ...@@ -305,7 +308,7 @@ Transport::handleIncoming(std::shared_ptr<Connection> connection) {
ssize_t totalBytes = 0; ssize_t totalBytes = 0;
for (;;) { for (;;) {
ssize_t bytes = recv(connection->fd, buffer + totalBytes, Const::MaxBuffer - totalBytes, 0); ssize_t bytes = recv(connection->fd(), buffer + totalBytes, Const::MaxBuffer - totalBytes, 0);
if (bytes == -1) { if (bytes == -1) {
if (errno == EAGAIN || errno == EWOULDBLOCK) { if (errno == EAGAIN || errno == EWOULDBLOCK) {
if (totalBytes > 0) { if (totalBytes > 0) {
...@@ -322,7 +325,7 @@ Transport::handleIncoming(std::shared_ptr<Connection> connection) { ...@@ -322,7 +325,7 @@ Transport::handleIncoming(std::shared_ptr<Connection> connection) {
} else { } else {
connection->handleError("Remote closed connection"); connection->handleError("Remote closed connection");
} }
connections.erase(connection->fd); connections.erase(connection->fd());
connection->close(); connection->close();
break; break;
} }
...@@ -348,7 +351,7 @@ Transport::handleTimeout(const std::shared_ptr<Connection>& connection) { ...@@ -348,7 +351,7 @@ Transport::handleTimeout(const std::shared_ptr<Connection>& connection) {
} }
Connection::Connection() Connection::Connection()
: fd(-1) : fd_(-1)
, requestEntry(nullptr) , requestEntry(nullptr)
{ {
state_.store(static_cast<uint32_t>(State::Idle)); state_.store(static_cast<uint32_t>(State::Idle));
...@@ -382,7 +385,7 @@ Connection::connect(const Address& addr) ...@@ -382,7 +385,7 @@ Connection::connect(const Address& addr)
make_non_blocking(sfd); make_non_blocking(sfd);
connectionState_.store(Connecting); connectionState_.store(Connecting);
fd = sfd; fd_ = sfd;
transport_->asyncConnect(shared_from_this(), addr->ai_addr, addr->ai_addrlen) transport_->asyncConnect(shared_from_this(), addr->ai_addr, addr->ai_addrlen)
.then([=]() { .then([=]() {
...@@ -402,7 +405,7 @@ Connection::connect(const Address& addr) ...@@ -402,7 +405,7 @@ Connection::connect(const Address& addr)
std::string std::string
Connection::dump() const { Connection::dump() const {
std::ostringstream oss; std::ostringstream oss;
oss << "Connection(fd = " << fd << ", src_port = "; oss << "Connection(fd = " << fd_ << ", src_port = ";
oss << ntohs(saddr.sin_port) << ")"; oss << ntohs(saddr.sin_port) << ")";
return oss.str(); return oss.str();
} }
...@@ -420,7 +423,7 @@ Connection::isConnected() const { ...@@ -420,7 +423,7 @@ Connection::isConnected() const {
void void
Connection::close() { Connection::close() {
connectionState_.store(NotConnected); connectionState_.store(NotConnected);
::close(fd); ::close(fd_);
} }
void void
...@@ -436,6 +439,12 @@ Connection::hasTransport() const { ...@@ -436,6 +439,12 @@ Connection::hasTransport() const {
return transport_ != nullptr; return transport_ != nullptr;
} }
Fd Connection::fd() const
{
assert(fd_ != -1);
return fd_;
}
void void
Connection::handleResponsePacket(const char* buffer, size_t totalBytes) { Connection::handleResponsePacket(const char* buffer, size_t totalBytes) {
......
...@@ -4,19 +4,16 @@ ...@@ -4,19 +4,16 @@
Implementation of common HTTP headers described by the RFC Implementation of common HTTP headers described by the RFC
*/ */
#include <stdexcept>
#include <iterator>
#include <limits>
#include <cstring>
#include <iostream>
#include <pistache/http_header.h> #include <pistache/http_header.h>
#include <pistache/common.h> #include <pistache/common.h>
#include <pistache/http.h> #include <pistache/http.h>
#include <pistache/stream.h> #include <pistache/stream.h>
#include <arpa/inet.h>
using namespace std; #include <stdexcept>
#include <iterator>
#include <limits>
#include <cstring>
#include <iostream>
namespace Pistache { namespace Pistache {
namespace Http { namespace Http {
...@@ -340,53 +337,19 @@ Host::Host(const std::string& data) ...@@ -340,53 +337,19 @@ Host::Host(const std::string& data)
parse(data); parse(data);
} }
void void Host::parse(const std::string& data)
Host::parse(const std::string& data) { {
unsigned long pos = data.find(']'); AddressParser parser(data);
unsigned long s_pos = data.find('['); host_ = parser.rawHost();
if (pos != std::string::npos && s_pos != std::string::npos) { const std::string& port = parser.rawPort();
//IPv6 address if (port.empty())
host_ = data.substr(s_pos, pos + 1); {
try {
in6_addr addr6;
char buff6[INET6_ADDRSTRLEN + 1] = {0, };
std::copy(&host_[0], &host_[0] + host_.size(), buff6);
inet_pton(AF_INET6, buff6, &(addr6.s6_addr16));
} catch (std::runtime_error) {
throw std::invalid_argument("Invalid IPv6 address");
}
pos++;
} else {
//IPv4 address
pos = data.find(':');
if (pos == std::string::npos) {
host_ = data;
port_ = HTTP_STANDARD_PORT;
}
host_ = data.substr(0, pos);
if (host_ == "*") {
host_ = "0.0.0.0";
}
try {
in_addr addr;
char buff[INET_ADDRSTRLEN + 1] = {0, };
std::copy(&host_[0], &host_[0] + host_.size(), buff);
inet_pton(AF_INET, buff, &(addr));
} catch (std::runtime_error) {
throw std::invalid_argument("Invalid IPv4 address");
}
}
char *end;
const std::string portPart = data.substr(pos + 1);
long port;
if (pos != std::string::npos) {
port = strtol(portPart.c_str(), &end, 10);
if (port < std::numeric_limits<uint16_t>::min()|| port > std::numeric_limits<uint16_t>::max())
throw std::invalid_argument("Invalid port");
port_ = static_cast<uint16_t>(port);
} else {
port_ = HTTP_STANDARD_PORT; port_ = HTTP_STANDARD_PORT;
} }
else
{
port_ = Port(port);
}
} }
void void
......
...@@ -3,6 +3,9 @@ ...@@ -3,6 +3,9 @@
*/ */
#include <pistache/net.h>
#include <pistache/common.h>
#include <stdexcept> #include <stdexcept>
#include <limits> #include <limits>
#include <cstring> #include <cstring>
...@@ -12,16 +15,42 @@ ...@@ -12,16 +15,42 @@
#include <ifaddrs.h> #include <ifaddrs.h>
#include <iostream> #include <iostream>
#include <pistache/net.h> namespace Pistache {
#include <pistache/common.h>
using namespace std; namespace {
bool IsIPv4HostName(const std::string& host)
{
in_addr addr;
char buff[INET_ADDRSTRLEN + 1] = {0, };
std::copy(host.begin(), host.end(), buff);
int res = inet_pton(AF_INET, buff, &addr);
return res;
}
namespace Pistache { bool IsIPv6HostName(const std::string& host)
{
in6_addr addr6;
char buff6[INET6_ADDRSTRLEN + 1] = {0, };
std::copy(host.begin(), host.end(), buff6);
int res = inet_pton(AF_INET6, buff6, &(addr6.s6_addr16));
return res;
}
}
Port::Port(uint16_t port) Port::Port(uint16_t port)
: port(port) : port(port)
{ } {}
Port::Port(const std::string& data)
{
if (data.empty())
throw std::invalid_argument("Invalid port: empty port");
char *end = 0;
long port_num = strtol(data.c_str(), &end, 10);
if (*end != 0 || port_num < Port::min() || port_num > Port::max())
throw std::invalid_argument("Invalid port: " + data);
port = static_cast<uint16_t>(port_num);
}
bool bool
Port::isReserved() const { Port::isReserved() const {
...@@ -157,6 +186,48 @@ bool Ipv6::supported() { ...@@ -157,6 +186,48 @@ bool Ipv6::supported() {
return supportsIpv6; return supportsIpv6;
} }
AddressParser::AddressParser(const std::string& data)
{
std::size_t end_pos = data.find(']');
std::size_t start_pos = data.find('[');
if (start_pos != std::string::npos &&
end_pos != std::string::npos &&
start_pos < end_pos)
{
host_ = data.substr(start_pos, end_pos + 1);
family_ = AF_INET6;
++end_pos;
}
else
{
end_pos = data.find(':');
host_ = data.substr(0, end_pos);
family_ = AF_INET;
}
if (end_pos != std::string::npos)
{
port_ = data.substr(end_pos + 1);
if (port_.empty())
throw std::invalid_argument("Invalid port");
}
}
const std::string& AddressParser::rawHost() const
{
return host_;
}
const std::string& AddressParser::rawPort() const
{
return port_;
}
int AddressParser::family() const
{
return family_;
}
Address::Address() Address::Address()
: host_("") : host_("")
, port_(0) , port_(0)
...@@ -228,46 +299,49 @@ Address::family() const { ...@@ -228,46 +299,49 @@ Address::family() const {
return family_; return family_;
} }
void void Address::init(const std::string& addr)
Address::init(const std::string& addr) { {
unsigned long pos = addr.find(']'); AddressParser parser(addr);
unsigned long s_pos = addr.find('['); family_ = parser.family();
if (pos != std::string::npos && s_pos != std::string::npos) { if (family_ == AF_INET6)
//IPv6 address {
host_ = addr.substr(s_pos+1, pos-1); const std::string& raw_host = parser.rawHost();
family_ = AF_INET6; assert(raw_host.size() > 2);
try { host_ = addr.substr(1, raw_host.size() - 2);
in6_addr addr6;
inet_pton(AF_INET6, host_.c_str(), &(addr6.s6_addr16)); if (!IsIPv6HostName(host_))
} catch (std::runtime_error) { {
throw std::invalid_argument("Invalid IPv6 address"); throw std::invalid_argument("Invalid IPv6 address");
} }
pos++; }
} else { else if (family_ == AF_INET)
//IPv4 address {
pos = addr.find(':'); host_ = parser.rawHost();
if (pos == std::string::npos) if (host_ == "*")
throw std::invalid_argument("Invalid address"); {
host_ = addr.substr(0, pos);
family_ = AF_INET;
if (host_ == "*") {
host_ = "0.0.0.0"; host_ = "0.0.0.0";
} }
try { else if (host_ == "localhost")
in_addr addr; {
inet_pton(AF_INET, host_.c_str(), &(addr)); host_ = "127.0.0.1";
} catch (std::runtime_error) { }
if (!IsIPv4HostName(host_))
{
throw std::invalid_argument("Invalid IPv4 address"); throw std::invalid_argument("Invalid IPv4 address");
} }
} }
char *end; else
const std::string portPart = addr.substr(pos + 1); { assert(false); }
const std::string& portPart = parser.rawPort();
if (portPart.empty()) if (portPart.empty())
throw std::invalid_argument("Invalid port"); throw std::invalid_argument("Invalid port");
char *end = 0;
long port = strtol(portPart.c_str(), &end, 10); long port = strtol(portPart.c_str(), &end, 10);
if (*end != 0 || port < Port::min() || port > Port::max()) if (*end != 0 || port < Port::min() || port > Port::max())
throw std::invalid_argument("Invalid port"); throw std::invalid_argument("Invalid port");
port_ = static_cast<uint16_t>(port); port_ = Port(port);
} }
Error::Error(const char* message) Error::Error(const char* message)
......
...@@ -30,13 +30,13 @@ Peer::Peer(const Address& addr) ...@@ -30,13 +30,13 @@ Peer::Peer(const Address& addr)
, ssl_(NULL) , ssl_(NULL)
{ } { }
Address const Address& Peer::address() const
Peer::address() const { {
return addr; return addr;
} }
string const std::string& Peer::hostname() const
Peer::hostname() const { {
return hostname_; return hostname_;
} }
......
...@@ -3,6 +3,8 @@ ...@@ -3,6 +3,8 @@
*/ */
#include <pistache/stream.h>
#include <iostream> #include <iostream>
#include <algorithm> #include <algorithm>
#include <string> #include <string>
...@@ -12,10 +14,43 @@ ...@@ -12,10 +14,43 @@
#include <unistd.h> #include <unistd.h>
#include <fcntl.h> #include <fcntl.h>
#include <pistache/stream.h>
namespace Pistache { namespace Pistache {
Buffer::Buffer()
: data()
, length(0)
, isDetached(false)
{ }
Buffer::Buffer(std::string data, int length, bool isDetached)
: data(std::move(data))
, length(length)
, isDetached(isDetached)
{ }
Buffer::Buffer(const char* data, int length, bool isDetached)
: data()
, length(length)
, isDetached(isDetached)
{
this->data.resize(length + 1);
this->data.assign(data, length + 1);
}
Buffer Buffer::detach(size_t fromIndex)
{
if (data.empty())
return Buffer();
if (length < fromIndex)
throw std::range_error("Trying to detach buffer from an index bigger than lengthght.");
auto newDatalength = length - fromIndex;
std::string newData = data.substr(fromIndex, newDatalength);
return Buffer(std::move(newData), newDatalength, true);
}
FileBuffer::FileBuffer(const std::string& fileName) FileBuffer::FileBuffer(const std::string& fileName)
: fileName_(fileName) : fileName_(fileName)
......
...@@ -192,12 +192,6 @@ Transport::handlePeerDisconnection(const std::shared_ptr<Peer>& peer) { ...@@ -192,12 +192,6 @@ Transport::handlePeerDisconnection(const std::shared_ptr<Peer>& peer) {
Guard guard(toWriteLock); Guard guard(toWriteLock);
auto & wq = toWrite[fd]; auto & wq = toWrite[fd];
while (wq.size() > 0) { while (wq.size() > 0) {
auto & entry = wq.front();
const BufferHolder & buffer = entry.buffer;
if (buffer.isRaw()) {
auto raw = buffer.raw();
if (raw.isOwned) delete[] raw.data;
}
wq.pop_front(); wq.pop_front();
} }
toWrite.erase(fd); toWrite.erase(fd);
...@@ -224,14 +218,10 @@ Transport::asyncWriteImpl(Fd fd) ...@@ -224,14 +218,10 @@ Transport::asyncWriteImpl(Fd fd)
auto & entry = wq.front(); auto & entry = wq.front();
int flags = entry.flags; int flags = entry.flags;
const BufferHolder &buffer = entry.buffer; BufferHolder &buffer = entry.buffer;
Async::Deferred<ssize_t> deferred = std::move(entry.deferred); Async::Deferred<ssize_t> deferred = std::move(entry.deferred);
auto cleanUp = [&]() { auto cleanUp = [&]() {
if (buffer.isRaw()) {
auto raw = buffer.raw();
if (raw.isOwned) delete[] raw.data;
}
wq.pop_front(); wq.pop_front();
if (wq.size() == 0) { if (wq.size() == 0) {
toWrite.erase(fd); toWrite.erase(fd);
...@@ -247,7 +237,7 @@ Transport::asyncWriteImpl(Fd fd) ...@@ -247,7 +237,7 @@ Transport::asyncWriteImpl(Fd fd)
if (buffer.isRaw()) { if (buffer.isRaw()) {
auto raw = buffer.raw(); auto raw = buffer.raw();
auto ptr = raw.data + totalWritten; auto ptr = raw.data.c_str() + totalWritten;
#ifdef PISTACHE_USE_SSL #ifdef PISTACHE_USE_SSL
auto it = peers.find(fd); auto it = peers.find(fd);
......
...@@ -28,6 +28,7 @@ pistache_test(streaming_test) ...@@ -28,6 +28,7 @@ pistache_test(streaming_test)
pistache_test(rest_server_test) pistache_test(rest_server_test)
pistache_test(string_view_test) pistache_test(string_view_test)
pistache_test(mailbox_test) pistache_test(mailbox_test)
pistache_test(stream_test)
if (PISTACHE_SSL) if (PISTACHE_SSL)
......
...@@ -256,6 +256,10 @@ TEST(headers_test, host) { ...@@ -256,6 +256,10 @@ TEST(headers_test, host) {
ASSERT_EQ(host.host(), "www.w3.org"); ASSERT_EQ(host.host(), "www.w3.org");
ASSERT_EQ(host.port(), 80); ASSERT_EQ(host.port(), 80);
host.parse("www.example.com:8080");
ASSERT_EQ(host.host(), "www.example.com");
ASSERT_EQ(host.port(), 8080);
host.parse("localhost:8080"); host.parse("localhost:8080");
ASSERT_EQ(host.host(), "localhost"); ASSERT_EQ(host.host(), "localhost");
ASSERT_EQ(host.port(), 8080); ASSERT_EQ(host.port(), 8080);
......
...@@ -90,9 +90,24 @@ TEST(net_test, invalid_address) ...@@ -90,9 +90,24 @@ TEST(net_test, invalid_address)
ASSERT_THROW(Address("127.0.0.1:"), std::invalid_argument); ASSERT_THROW(Address("127.0.0.1:"), std::invalid_argument);
ASSERT_THROW(Address("127.0.0.1:-10"), std::invalid_argument); ASSERT_THROW(Address("127.0.0.1:-10"), std::invalid_argument);
/* Due to an error in GLIBC these tests don't fail as expected, further research needed */ ASSERT_THROW(Address("[GGGG:GGGG:GGGG:GGGG:GGGG:GGGG:GGGG:GGGG]:8080");, std::invalid_argument);
// ASSERT_THROW(Address("[GGGG:GGGG:GGGG:GGGG:GGGG:GGGG:GGGG:GGGG]:8080");, std::invalid_argument); ASSERT_THROW(Address("[::GGGG]:8080");, std::invalid_argument);
// ASSERT_THROW(Address("[::GGGG]:8080");, std::invalid_argument); ASSERT_THROW(Address("256.256.256.256:8080");, std::invalid_argument);
// ASSERT_THROW(Address("256.256.256.256:8080");, std::invalid_argument); ASSERT_THROW(Address("1.0.0.256:8080");, std::invalid_argument);
// ASSERT_THROW(Address("1.0.0.256:8080");, std::invalid_argument); }
TEST(net_test, address_parser)
{
AddressParser ap1("127.0.0.1:80");
ASSERT_EQ(ap1.rawHost(), "127.0.0.1");
ASSERT_EQ(ap1.rawPort(), "80");
ASSERT_EQ(ap1.family(), AF_INET);
AddressParser ap2("[2001:0DB8:AABB:CCDD:EEFF:0011:2233:4455]:8080");
ASSERT_EQ(ap2.rawHost(), "[2001:0DB8:AABB:CCDD:EEFF:0011:2233:4455]");
ASSERT_EQ(ap2.rawPort(), "8080");
ASSERT_EQ(ap2.family(), AF_INET6);
ASSERT_THROW(AddressParser("127.0.0.1:");, std::invalid_argument);
ASSERT_THROW(AddressParser("[::]:");, std::invalid_argument);
} }
#include <pistache/stream.h>
#include "gtest/gtest.h"
#include <cstring>
using namespace Pistache;
TEST(stream, test_buffer)
{
char str[] = "test_string";
size_t len = strlen(str);
Buffer buffer1(str, len, false);
Buffer buffer2 = buffer1.detach(0);
ASSERT_EQ(buffer2.length, len);
ASSERT_EQ(buffer2.isDetached, true);
Buffer buffer3;
ASSERT_EQ(buffer3.length, 0u);
ASSERT_EQ(buffer3.isDetached, false);
Buffer buffer4 = buffer3.detach(0);
ASSERT_EQ(buffer4.length, 0u);
ASSERT_EQ(buffer4.isDetached, false);
ASSERT_THROW(buffer1.detach(2 * len);, std::range_error);
}
\ No newline at end of file
...@@ -46,7 +46,7 @@ void dumpData(const Rest::Request&req, Http::ResponseWriter response) { ...@@ -46,7 +46,7 @@ void dumpData(const Rest::Request&req, Http::ResponseWriter response) {
stream.ends(); stream.ends();
} }
TEST(stream, from_description) TEST(streaming, from_description)
{ {
Address addr(Ipv4::any(), Port(0)); Address addr(Ipv4::any(), Port(0));
const size_t threads = 20; const size_t threads = 20;
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment