mirror of
https://github.com/apache/thrift.git
synced 2026-10-29 17:05:46 +08:00
425 lines
14 KiB
C++
425 lines
14 KiB
C++
/*
|
|
* Licensed to the Apache Software Foundation (ASF) under one
|
|
* or more contributor license agreements. See the NOTICE file
|
|
* distributed with this work for additional information
|
|
* regarding copyright ownership. The ASF licenses this file
|
|
* to you under the Apache License, Version 2.0 (the
|
|
* "License"); you may not use this file except in compliance
|
|
* with the License. You may obtain a copy of the License at
|
|
*
|
|
* http://www.apache.org/licenses/LICENSE-2.0
|
|
*
|
|
* Unless required by applicable law or agreed to in writing,
|
|
* software distributed under the License is distributed on an
|
|
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
|
* KIND, either express or implied. See the License for the
|
|
* specific language governing permissions and limitations
|
|
* under the License.
|
|
*/
|
|
|
|
#include "gen-cpp/OneWayService.h"
|
|
#include <boost/test/unit_test.hpp>
|
|
#include <boost/thread.hpp>
|
|
#include <climits>
|
|
#include <iostream>
|
|
#include <memory>
|
|
#include <sstream>
|
|
#include <thrift/concurrency/Monitor.h>
|
|
#include <thrift/protocol/TBinaryProtocol.h>
|
|
#include <thrift/protocol/TJSONProtocol.h>
|
|
#include <thrift/server/TThreadedServer.h>
|
|
#include <thrift/transport/TBufferTransports.h>
|
|
#include <thrift/transport/THttpClient.h>
|
|
#include <thrift/transport/THttpServer.h>
|
|
#include <thrift/transport/TServerSocket.h>
|
|
#include <thrift/transport/TSocket.h>
|
|
#include <vector>
|
|
|
|
BOOST_AUTO_TEST_SUITE(OneWayHTTPTest)
|
|
|
|
using namespace apache::thrift;
|
|
using apache::thrift::protocol::TProtocol;
|
|
using apache::thrift::protocol::TBinaryProtocol;
|
|
using apache::thrift::protocol::TBinaryProtocolFactory;
|
|
using apache::thrift::protocol::TJSONProtocol;
|
|
using apache::thrift::protocol::TJSONProtocolFactory;
|
|
using apache::thrift::server::TThreadedServer;
|
|
using apache::thrift::server::TServerEventHandler;
|
|
using apache::thrift::transport::TTransport;
|
|
using apache::thrift::transport::THttpServer;
|
|
using apache::thrift::transport::THttpServerTransportFactory;
|
|
using apache::thrift::transport::THttpClient;
|
|
using apache::thrift::transport::TBufferedTransport;
|
|
using apache::thrift::transport::TBufferedTransportFactory;
|
|
using apache::thrift::transport::TMemoryBuffer;
|
|
using apache::thrift::transport::TServerSocket;
|
|
using apache::thrift::transport::TSocket;
|
|
using apache::thrift::transport::TTransportException;
|
|
using std::shared_ptr;
|
|
using std::string;
|
|
namespace utf = boost::unit_test;
|
|
|
|
// Define this env var to enable some logging (in case you need to debug)
|
|
#undef ENABLE_STDERR_LOGGING
|
|
|
|
class TInspectableHttpClient : public THttpClient {
|
|
public:
|
|
explicit TInspectableHttpClient(std::shared_ptr<TTransport> transport) : THttpClient(transport) {}
|
|
|
|
bool closesAfterHeader(const string& header) {
|
|
closeAfterResponse_ = false;
|
|
std::vector<char> buffer(header.begin(), header.end());
|
|
buffer.push_back('\0');
|
|
parseHeader(buffer.data());
|
|
return closeAfterResponse_;
|
|
}
|
|
|
|
bool chunksAfterHeader(const string& header) {
|
|
chunked_ = false;
|
|
std::vector<char> buffer(header.begin(), header.end());
|
|
buffer.push_back('\0');
|
|
parseHeader(buffer.data());
|
|
return chunked_;
|
|
}
|
|
|
|
uint32_t contentLengthAfterHeader(const string& header) {
|
|
contentLength_ = 0;
|
|
std::vector<char> buffer(header.begin(), header.end());
|
|
buffer.push_back('\0');
|
|
parseHeader(buffer.data());
|
|
return contentLength_;
|
|
}
|
|
};
|
|
|
|
class OneWayServiceHandler : public onewaytest::OneWayServiceIf {
|
|
public:
|
|
OneWayServiceHandler() = default;
|
|
|
|
void roundTripRPC() override {
|
|
#ifdef ENABLE_STDERR_LOGGING
|
|
cerr << "roundTripRPC()" << '\n';
|
|
#endif
|
|
}
|
|
void oneWayRPC() override {
|
|
#ifdef ENABLE_STDERR_LOGGING
|
|
cerr << "oneWayRPC()" << '\n';
|
|
#endif
|
|
}
|
|
};
|
|
|
|
class OneWayServiceCloneFactory : virtual public onewaytest::OneWayServiceIfFactory {
|
|
public:
|
|
~OneWayServiceCloneFactory() override = default;
|
|
onewaytest::OneWayServiceIf* getHandler(const ::apache::thrift::TConnectionInfo& connInfo) override
|
|
{
|
|
(void)connInfo ;
|
|
return new OneWayServiceHandler;
|
|
}
|
|
void releaseHandler( onewaytest::OneWayServiceIf* handler) override {
|
|
delete handler;
|
|
}
|
|
};
|
|
|
|
class RPC0ThreadClass {
|
|
public:
|
|
RPC0ThreadClass(TThreadedServer& server) : server_(server) { } // Constructor
|
|
~RPC0ThreadClass() = default; // Destructor
|
|
|
|
void Run() {
|
|
server_.serve() ;
|
|
}
|
|
TThreadedServer& server_ ;
|
|
} ;
|
|
|
|
using apache::thrift::concurrency::Monitor;
|
|
using apache::thrift::concurrency::Mutex;
|
|
using apache::thrift::concurrency::Synchronized;
|
|
|
|
// copied from IntegrationTest
|
|
class TServerReadyEventHandler : public TServerEventHandler, public Monitor {
|
|
public:
|
|
TServerReadyEventHandler() : isListening_(false), accepted_(0) {}
|
|
~TServerReadyEventHandler() override = default;
|
|
void preServe() override {
|
|
Synchronized sync(*this);
|
|
isListening_ = true;
|
|
notify();
|
|
}
|
|
void* createContext(shared_ptr<TProtocol> input,
|
|
shared_ptr<TProtocol> output) override {
|
|
Synchronized sync(*this);
|
|
++accepted_;
|
|
notify();
|
|
|
|
(void)input;
|
|
(void)output;
|
|
return nullptr;
|
|
}
|
|
bool isListening() const { return isListening_; }
|
|
uint64_t acceptedCount() {
|
|
Synchronized sync(*this);
|
|
return accepted_;
|
|
}
|
|
|
|
private:
|
|
bool isListening_;
|
|
uint64_t accepted_;
|
|
};
|
|
|
|
class TClosingHttpServer : public THttpServer {
|
|
public:
|
|
explicit TClosingHttpServer(std::shared_ptr<TTransport> transport)
|
|
: THttpServer(transport, transport->getConfiguration()) {}
|
|
|
|
void flush() override {
|
|
resetConsumedMessageSize();
|
|
|
|
uint8_t* buf;
|
|
uint32_t len;
|
|
writeBuffer_.getBuffer(&buf, &len);
|
|
|
|
std::ostringstream header;
|
|
header << "HTTP/1.1 200 OK\r\n"
|
|
<< "Content-Type: application/x-thrift\r\n"
|
|
<< "Transfer-Encoding: chunked\r\n"
|
|
<< "Connection: keep-alive, close\r\n"
|
|
<< "Connection: keep-alive\r\n\r\n";
|
|
const string headerText = header.str();
|
|
transport_->write(reinterpret_cast<const uint8_t*>(headerText.data()),
|
|
static_cast<uint32_t>(headerText.size()));
|
|
|
|
if (len > 0) {
|
|
std::ostringstream chunkSize;
|
|
chunkSize << std::hex << len << "\r\n";
|
|
const string chunkPrefix = chunkSize.str();
|
|
transport_->write(reinterpret_cast<const uint8_t*>(chunkPrefix.data()),
|
|
static_cast<uint32_t>(chunkPrefix.size()));
|
|
transport_->write(buf, len);
|
|
transport_->write(reinterpret_cast<const uint8_t*>("\r\n"), 2);
|
|
}
|
|
transport_->write(reinterpret_cast<const uint8_t*>("0\r\n\r\n"), 5);
|
|
transport_->flush();
|
|
|
|
writeBuffer_.resetBuffer();
|
|
readHeaders_ = true;
|
|
close();
|
|
}
|
|
};
|
|
|
|
class TClosingHttpServerTransportFactory : public apache::thrift::transport::TTransportFactory {
|
|
public:
|
|
std::shared_ptr<TTransport> getTransport(std::shared_ptr<TTransport> transport) override {
|
|
return std::make_shared<TClosingHttpServer>(transport);
|
|
}
|
|
};
|
|
|
|
class TBlockableBufferedTransport : public TBufferedTransport {
|
|
public:
|
|
TBlockableBufferedTransport(std::shared_ptr<TTransport> transport)
|
|
: TBufferedTransport(transport, 10240),
|
|
blocked_(false) {
|
|
}
|
|
|
|
uint32_t write_buffer_length() {
|
|
auto have_bytes = static_cast<uint32_t>(wBase_ - wBuf_.get());
|
|
return have_bytes ;
|
|
}
|
|
|
|
void block() {
|
|
blocked_ = true ;
|
|
#ifdef ENABLE_STDERR_LOGGING
|
|
cerr << "block flushing\n" ;
|
|
#endif
|
|
}
|
|
void unblock() {
|
|
blocked_ = false ;
|
|
#ifdef ENABLE_STDERR_LOGGING
|
|
cerr << "unblock flushing, buffer is\n<<" << std::string((char *)wBuf_.get(), write_buffer_length()) << ">>\n" ;
|
|
#endif
|
|
}
|
|
|
|
void flush() override {
|
|
if (blocked_) {
|
|
#ifdef ENABLE_STDERR_LOGGING
|
|
cerr << "flush was blocked\n" ;
|
|
#endif
|
|
return ;
|
|
}
|
|
TBufferedTransport::flush() ;
|
|
}
|
|
|
|
bool blocked_ ;
|
|
} ;
|
|
|
|
BOOST_AUTO_TEST_CASE( JSON_BufferedHTTP )
|
|
{
|
|
std::shared_ptr<TServerSocket> ss = std::make_shared<TServerSocket>(0) ;
|
|
TThreadedServer server(
|
|
std::make_shared<onewaytest::OneWayServiceProcessorFactory>(std::make_shared<OneWayServiceCloneFactory>()),
|
|
ss, //port
|
|
std::make_shared<THttpServerTransportFactory>(),
|
|
std::make_shared<TJSONProtocolFactory>());
|
|
|
|
std::shared_ptr<TServerReadyEventHandler> pEventHandler(new TServerReadyEventHandler) ;
|
|
server.setServerEventHandler(pEventHandler);
|
|
|
|
#ifdef ENABLE_STDERR_LOGGING
|
|
cerr << "Starting the server...\n";
|
|
#endif
|
|
RPC0ThreadClass t(server) ;
|
|
boost::thread thread(&RPC0ThreadClass::Run, &t);
|
|
|
|
{
|
|
Synchronized sync(*(pEventHandler.get()));
|
|
while (!pEventHandler->isListening()) {
|
|
pEventHandler->wait();
|
|
}
|
|
}
|
|
|
|
int port = ss->getPort() ;
|
|
#ifdef ENABLE_STDERR_LOGGING
|
|
cerr << "port " << port << '\n';
|
|
#endif
|
|
|
|
{
|
|
std::shared_ptr<TSocket> socket(new TSocket("localhost", port));
|
|
socket->setRecvTimeout(10000) ; // 1000msec should be enough
|
|
std::shared_ptr<TBlockableBufferedTransport> blockable_transport(new TBlockableBufferedTransport(socket));
|
|
std::shared_ptr<TTransport> transport(new THttpClient(blockable_transport, "localhost", "/service"));
|
|
std::shared_ptr<TProtocol> protocol(new TJSONProtocol(transport));
|
|
onewaytest::OneWayServiceClient client(protocol);
|
|
|
|
|
|
transport->open();
|
|
client.roundTripRPC();
|
|
blockable_transport->block() ;
|
|
uint32_t size0 = blockable_transport->write_buffer_length() ;
|
|
client.send_oneWayRPC() ;
|
|
uint32_t size1 = blockable_transport->write_buffer_length() ;
|
|
client.send_oneWayRPC() ;
|
|
uint32_t size2 = blockable_transport->write_buffer_length() ;
|
|
BOOST_CHECK((size1 - size0) == (size2 - size1)) ;
|
|
blockable_transport->unblock() ;
|
|
client.send_roundTripRPC() ;
|
|
blockable_transport->flush() ;
|
|
uint8_t discard;
|
|
BOOST_CHECK_EQUAL(transport->read(&discard, 1), 0U);
|
|
BOOST_CHECK_EQUAL(transport->read(&discard, 1), 0U);
|
|
try {
|
|
client.recv_roundTripRPC() ;
|
|
} catch (const TTransportException &e) {
|
|
BOOST_ERROR( "we should not get a transport exception -- this means we failed: " + std::string(e.what()) ) ;
|
|
}
|
|
transport->close();
|
|
}
|
|
server.stop();
|
|
thread.join() ;
|
|
#ifdef ENABLE_STDERR_LOGGING
|
|
cerr << "finished.\n";
|
|
#endif
|
|
}
|
|
|
|
BOOST_AUTO_TEST_CASE( JSON_HTTP_OneWayWrapperDoesNotPoisonNextCall )
|
|
{
|
|
std::shared_ptr<TServerSocket> ss = std::make_shared<TServerSocket>(0);
|
|
TThreadedServer server(
|
|
std::make_shared<onewaytest::OneWayServiceProcessorFactory>(
|
|
std::make_shared<OneWayServiceCloneFactory>()),
|
|
ss,
|
|
std::make_shared<THttpServerTransportFactory>(),
|
|
std::make_shared<TJSONProtocolFactory>());
|
|
|
|
std::shared_ptr<TServerReadyEventHandler> pEventHandler(new TServerReadyEventHandler);
|
|
server.setServerEventHandler(pEventHandler);
|
|
|
|
RPC0ThreadClass t(server);
|
|
boost::thread thread(&RPC0ThreadClass::Run, &t);
|
|
|
|
{
|
|
Synchronized sync(*(pEventHandler.get()));
|
|
while (!pEventHandler->isListening()) {
|
|
pEventHandler->wait();
|
|
}
|
|
}
|
|
|
|
{
|
|
std::shared_ptr<TSocket> socket(new TSocket("localhost", ss->getPort()));
|
|
socket->setRecvTimeout(10000);
|
|
std::shared_ptr<TTransport> transport(new THttpClient(socket, "localhost", "/service"));
|
|
std::shared_ptr<TProtocol> protocol(new TJSONProtocol(transport));
|
|
onewaytest::OneWayServiceClient client(protocol);
|
|
|
|
transport->open();
|
|
client.roundTripRPC();
|
|
BOOST_CHECK_EQUAL(pEventHandler->acceptedCount(), 1U);
|
|
client.oneWayRPC();
|
|
try {
|
|
client.roundTripRPC();
|
|
} catch (const std::exception& e) {
|
|
BOOST_ERROR("roundTripRPC after oneWayRPC failed: " + std::string(e.what()));
|
|
}
|
|
BOOST_CHECK_EQUAL(pEventHandler->acceptedCount(), 1U);
|
|
transport->close();
|
|
}
|
|
|
|
server.stop();
|
|
thread.join();
|
|
}
|
|
|
|
BOOST_AUTO_TEST_CASE(HTTP_ClientReconnectsAfterConnectionClose) {
|
|
std::shared_ptr<TServerSocket> ss = std::make_shared<TServerSocket>(0);
|
|
TThreadedServer server(std::make_shared<onewaytest::OneWayServiceProcessorFactory>(
|
|
std::make_shared<OneWayServiceCloneFactory>()),
|
|
ss, std::make_shared<TClosingHttpServerTransportFactory>(),
|
|
std::make_shared<TBinaryProtocolFactory>());
|
|
|
|
std::shared_ptr<TServerReadyEventHandler> pEventHandler(new TServerReadyEventHandler);
|
|
server.setServerEventHandler(pEventHandler);
|
|
|
|
RPC0ThreadClass t(server);
|
|
boost::thread thread(&RPC0ThreadClass::Run, &t);
|
|
|
|
{
|
|
Synchronized sync(*(pEventHandler.get()));
|
|
while (!pEventHandler->isListening()) {
|
|
pEventHandler->wait();
|
|
}
|
|
}
|
|
|
|
{
|
|
std::shared_ptr<TSocket> socket(new TSocket("localhost", ss->getPort()));
|
|
socket->setRecvTimeout(10000);
|
|
std::shared_ptr<TTransport> httpTransport(new THttpClient(socket, "localhost", "/service"));
|
|
std::shared_ptr<TTransport> transport(new TBufferedTransport(httpTransport));
|
|
std::shared_ptr<TProtocol> protocol(new TBinaryProtocol(transport));
|
|
onewaytest::OneWayServiceClient client(protocol);
|
|
|
|
transport->open();
|
|
client.roundTripRPC();
|
|
BOOST_CHECK_EQUAL(pEventHandler->acceptedCount(), 1U);
|
|
BOOST_CHECK_NO_THROW(client.roundTripRPC());
|
|
BOOST_CHECK_EQUAL(pEventHandler->acceptedCount(), 2U);
|
|
client.oneWayRPC();
|
|
BOOST_CHECK_NO_THROW(client.roundTripRPC());
|
|
BOOST_CHECK_EQUAL(pEventHandler->acceptedCount(), 4U);
|
|
transport->close();
|
|
BOOST_CHECK_EQUAL(pEventHandler->acceptedCount(), 4U);
|
|
}
|
|
|
|
server.stop();
|
|
thread.join();
|
|
}
|
|
|
|
BOOST_AUTO_TEST_CASE(HTTP_ClientRequiresExactConnectionHeaderName) {
|
|
TInspectableHttpClient client(std::make_shared<TMemoryBuffer>());
|
|
|
|
BOOST_CHECK(client.closesAfterHeader("Connection: keep-alive, close"));
|
|
BOOST_CHECK(!client.closesAfterHeader("Connection-Timeout: close"));
|
|
BOOST_CHECK(client.chunksAfterHeader("Transfer-Encoding: chunked"));
|
|
BOOST_CHECK(!client.chunksAfterHeader("Transfer-Encoding-Other: chunked"));
|
|
BOOST_CHECK_EQUAL(client.contentLengthAfterHeader("Content-Length: 42"), 42U);
|
|
BOOST_CHECK_EQUAL(client.contentLengthAfterHeader("Content-Length-Mismatch: 42"), 0U);
|
|
}
|
|
|
|
BOOST_AUTO_TEST_SUITE_END()
|