mirror of
https://github.com/clearlinux/folly.git
synced 2026-09-01 02:34:43 +00:00
strip Channel from all class names
Summary: as above. Only got a little messy when components within folly::wangle typedefed things to Pipeline Test Plan: unit tests Reviewed By: davejwatson@fb.com Subscribers: wormhole-diffs@, fugalh, alandau, bmatheny, folly-diffs@, jsedgwick, yfeldblum, chalfant FB internal diff: D2022181 Tasks: 6836580 Signature: t1:2022181:1430157032:df0bdfb9ca0d76b86d52c55c4ad41ea953a18cb4
This commit is contained in:
committed by
Alecs King
parent
da10a8e64a
commit
3444ffba56
+3
-3
@@ -273,10 +273,10 @@ nobase_follyinclude_HEADERS = \
|
||||
wangle/bootstrap/ServerSocketFactory.h \
|
||||
wangle/bootstrap/ClientBootstrap.h \
|
||||
wangle/channel/AsyncSocketHandler.h \
|
||||
wangle/channel/ChannelHandler.h \
|
||||
wangle/channel/ChannelHandlerContext.h \
|
||||
wangle/channel/ChannelPipeline.h \
|
||||
wangle/channel/Handler.h \
|
||||
wangle/channel/HandlerContext.h \
|
||||
wangle/channel/OutputBufferingHandler.h \
|
||||
wangle/channel/Pipeline.h \
|
||||
wangle/concurrent/BlockingQueue.h \
|
||||
wangle/concurrent/Codel.h \
|
||||
wangle/concurrent/CPUThreadPoolExecutor.h \
|
||||
|
||||
@@ -16,7 +16,7 @@
|
||||
|
||||
#include "folly/wangle/bootstrap/ServerBootstrap.h"
|
||||
#include "folly/wangle/bootstrap/ClientBootstrap.h"
|
||||
#include "folly/wangle/channel/ChannelHandler.h"
|
||||
#include "folly/wangle/channel/Handler.h"
|
||||
|
||||
#include <glog/logging.h>
|
||||
#include <gtest/gtest.h>
|
||||
@@ -25,14 +25,14 @@
|
||||
using namespace folly::wangle;
|
||||
using namespace folly;
|
||||
|
||||
typedef ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> Pipeline;
|
||||
typedef Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> BytesPipeline;
|
||||
|
||||
typedef ServerBootstrap<Pipeline> TestServer;
|
||||
typedef ClientBootstrap<Pipeline> TestClient;
|
||||
typedef ServerBootstrap<BytesPipeline> TestServer;
|
||||
typedef ClientBootstrap<BytesPipeline> TestClient;
|
||||
|
||||
class TestClientPipelineFactory : public PipelineFactory<Pipeline> {
|
||||
class TestClientPipelineFactory : public PipelineFactory<BytesPipeline> {
|
||||
public:
|
||||
Pipeline* newPipeline(std::shared_ptr<AsyncSocket> sock) {
|
||||
BytesPipeline* newPipeline(std::shared_ptr<AsyncSocket> sock) {
|
||||
CHECK(sock->good());
|
||||
|
||||
// We probably aren't connected immedately, check after a small delay
|
||||
@@ -43,11 +43,11 @@ class TestClientPipelineFactory : public PipelineFactory<Pipeline> {
|
||||
}
|
||||
};
|
||||
|
||||
class TestPipelineFactory : public PipelineFactory<Pipeline> {
|
||||
class TestPipelineFactory : public PipelineFactory<BytesPipeline> {
|
||||
public:
|
||||
Pipeline* newPipeline(std::shared_ptr<AsyncSocket> sock) {
|
||||
BytesPipeline* newPipeline(std::shared_ptr<AsyncSocket> sock) {
|
||||
pipelines++;
|
||||
return new Pipeline();
|
||||
return new BytesPipeline();
|
||||
}
|
||||
std::atomic<int> pipelines{0};
|
||||
};
|
||||
@@ -268,7 +268,7 @@ TEST(Bootstrap, ExistingSocket) {
|
||||
std::atomic<int> connections{0};
|
||||
|
||||
class TestHandlerPipeline
|
||||
: public ChannelHandlerAdapter<void*,
|
||||
: public HandlerAdapter<void*,
|
||||
std::exception> {
|
||||
public:
|
||||
void read(Context* ctx, void* conn) {
|
||||
@@ -283,12 +283,12 @@ class TestHandlerPipeline
|
||||
|
||||
template <typename HandlerPipeline>
|
||||
class TestHandlerPipelineFactory
|
||||
: public PipelineFactory<ServerBootstrap<Pipeline>::AcceptPipeline> {
|
||||
: public PipelineFactory<ServerBootstrap<BytesPipeline>::AcceptPipeline> {
|
||||
public:
|
||||
ServerBootstrap<Pipeline>::AcceptPipeline* newPipeline(std::shared_ptr<AsyncSocket>) {
|
||||
auto pipeline = new ServerBootstrap<Pipeline>::AcceptPipeline;
|
||||
ServerBootstrap<BytesPipeline>::AcceptPipeline* newPipeline(std::shared_ptr<AsyncSocket>) {
|
||||
auto pipeline = new ServerBootstrap<BytesPipeline>::AcceptPipeline;
|
||||
auto handler = std::make_shared<HandlerPipeline>();
|
||||
pipeline->addBack(ChannelHandlerPtr<HandlerPipeline>(handler));
|
||||
pipeline->addBack(HandlerPtr<HandlerPipeline>(handler));
|
||||
return pipeline;
|
||||
}
|
||||
};
|
||||
@@ -318,7 +318,7 @@ TEST(Bootstrap, LoadBalanceHandler) {
|
||||
}
|
||||
|
||||
class TestUDPPipeline
|
||||
: public ChannelHandlerAdapter<void*,
|
||||
: public HandlerAdapter<void*,
|
||||
std::exception> {
|
||||
public:
|
||||
void read(Context* ctx, void* conn) {
|
||||
|
||||
@@ -15,12 +15,12 @@
|
||||
*/
|
||||
#pragma once
|
||||
|
||||
#include <folly/wangle/channel/ChannelPipeline.h>
|
||||
#include <folly/wangle/channel/Pipeline.h>
|
||||
|
||||
namespace folly {
|
||||
|
||||
/*
|
||||
* A thin wrapper around ChannelPipeline and AsyncSocket to match
|
||||
* A thin wrapper around Pipeline and AsyncSocket to match
|
||||
* ServerBootstrap. On connect() a new pipeline is created.
|
||||
*/
|
||||
template <typename Pipeline>
|
||||
|
||||
@@ -20,15 +20,15 @@
|
||||
#include <folly/io/async/EventBaseManager.h>
|
||||
#include <folly/wangle/concurrent/IOThreadPoolExecutor.h>
|
||||
#include <folly/wangle/acceptor/ManagedConnection.h>
|
||||
#include <folly/wangle/channel/ChannelPipeline.h>
|
||||
#include <folly/wangle/channel/ChannelHandler.h>
|
||||
#include <folly/wangle/channel/Pipeline.h>
|
||||
#include <folly/wangle/channel/Handler.h>
|
||||
|
||||
namespace folly {
|
||||
|
||||
template <typename Pipeline>
|
||||
class ServerAcceptor
|
||||
: public Acceptor
|
||||
, public folly::wangle::ChannelHandlerAdapter<void*, std::exception> {
|
||||
, public folly::wangle::HandlerAdapter<void*, std::exception> {
|
||||
typedef std::unique_ptr<Pipeline,
|
||||
folly::DelayedDestruction::Destructor> PipelinePtr;
|
||||
|
||||
@@ -60,7 +60,7 @@ class ServerAcceptor
|
||||
public:
|
||||
explicit ServerAcceptor(
|
||||
std::shared_ptr<PipelineFactory<Pipeline>> pipelineFactory,
|
||||
std::shared_ptr<folly::wangle::ChannelPipeline<
|
||||
std::shared_ptr<folly::wangle::Pipeline<
|
||||
void*, std::exception>> acceptorPipeline,
|
||||
EventBase* base)
|
||||
: Acceptor(ServerSocketConfig())
|
||||
@@ -70,7 +70,7 @@ class ServerAcceptor
|
||||
Acceptor::init(nullptr, base_);
|
||||
CHECK(acceptorPipeline_);
|
||||
|
||||
acceptorPipeline_->addBack(folly::wangle::ChannelHandlerPtr<ServerAcceptor, false>(this));
|
||||
acceptorPipeline_->addBack(folly::wangle::HandlerPtr<ServerAcceptor, false>(this));
|
||||
acceptorPipeline_->finalize();
|
||||
}
|
||||
|
||||
@@ -109,7 +109,7 @@ class ServerAcceptor
|
||||
EventBase* base_;
|
||||
|
||||
std::shared_ptr<PipelineFactory<Pipeline>> childPipelineFactory_;
|
||||
std::shared_ptr<folly::wangle::ChannelPipeline<
|
||||
std::shared_ptr<folly::wangle::Pipeline<
|
||||
void*, std::exception>> acceptorPipeline_;
|
||||
};
|
||||
|
||||
@@ -118,13 +118,13 @@ class ServerAcceptorFactory : public AcceptorFactory {
|
||||
public:
|
||||
explicit ServerAcceptorFactory(
|
||||
std::shared_ptr<PipelineFactory<Pipeline>> factory,
|
||||
std::shared_ptr<PipelineFactory<folly::wangle::ChannelPipeline<
|
||||
std::shared_ptr<PipelineFactory<folly::wangle::Pipeline<
|
||||
void*, std::exception>>> pipeline)
|
||||
: factory_(factory)
|
||||
, pipeline_(pipeline) {}
|
||||
|
||||
std::shared_ptr<Acceptor> newAcceptor(EventBase* base) {
|
||||
std::shared_ptr<folly::wangle::ChannelPipeline<
|
||||
std::shared_ptr<folly::wangle::Pipeline<
|
||||
void*, std::exception>> pipeline(
|
||||
pipeline_->newPipeline(nullptr));
|
||||
return std::make_shared<ServerAcceptor<Pipeline>>(factory_, pipeline, base);
|
||||
@@ -132,7 +132,7 @@ class ServerAcceptorFactory : public AcceptorFactory {
|
||||
private:
|
||||
std::shared_ptr<PipelineFactory<Pipeline>> factory_;
|
||||
std::shared_ptr<PipelineFactory<
|
||||
folly::wangle::ChannelPipeline<
|
||||
folly::wangle::Pipeline<
|
||||
void*, std::exception>>> pipeline_;
|
||||
};
|
||||
|
||||
@@ -183,8 +183,8 @@ void ServerWorkerPool::forEachWorker(F&& f) const {
|
||||
}
|
||||
|
||||
class DefaultAcceptPipelineFactory
|
||||
: public PipelineFactory<wangle::ChannelPipeline<void*, std::exception>> {
|
||||
typedef wangle::ChannelPipeline<
|
||||
: public PipelineFactory<wangle::Pipeline<void*, std::exception>> {
|
||||
typedef wangle::Pipeline<
|
||||
void*,
|
||||
std::exception> AcceptPipeline;
|
||||
|
||||
|
||||
@@ -15,7 +15,7 @@
|
||||
*/
|
||||
#include <folly/wangle/bootstrap/ServerBootstrap.h>
|
||||
#include <folly/wangle/concurrent/NamedThreadFactory.h>
|
||||
#include <folly/wangle/channel/ChannelHandler.h>
|
||||
#include <folly/wangle/channel/Handler.h>
|
||||
#include <folly/io/async/EventBaseManager.h>
|
||||
|
||||
namespace folly {
|
||||
|
||||
@@ -17,11 +17,11 @@
|
||||
|
||||
#include <folly/wangle/bootstrap/ServerBootstrap-inl.h>
|
||||
#include <folly/Baton.h>
|
||||
#include <folly/wangle/channel/ChannelPipeline.h>
|
||||
#include <folly/wangle/channel/Pipeline.h>
|
||||
|
||||
namespace folly {
|
||||
|
||||
typedef folly::wangle::ChannelPipeline<
|
||||
typedef folly::wangle::Pipeline<
|
||||
folly::IOBufQueue&, std::unique_ptr<folly::IOBuf>> DefaultPipeline;
|
||||
|
||||
/*
|
||||
@@ -30,7 +30,7 @@ typedef folly::wangle::ChannelPipeline<
|
||||
* accepting threads, any number of accepting sockets, a pool of
|
||||
* IO-worker threads, and connection pool for each IO thread for you.
|
||||
*
|
||||
* The output is given as a ChannelPipeline template: given a
|
||||
* The output is given as a Pipeline template: given a
|
||||
* PipelineFactory, it will create a new pipeline for each connection,
|
||||
* and your server can handle the incoming bytes.
|
||||
*
|
||||
@@ -52,7 +52,7 @@ class ServerBootstrap {
|
||||
join();
|
||||
}
|
||||
|
||||
typedef wangle::ChannelPipeline<
|
||||
typedef wangle::Pipeline<
|
||||
void*,
|
||||
std::exception> AcceptPipeline;
|
||||
/*
|
||||
|
||||
@@ -16,7 +16,7 @@
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <folly/wangle/channel/ChannelHandler.h>
|
||||
#include <folly/wangle/channel/Handler.h>
|
||||
#include <folly/io/async/AsyncSocket.h>
|
||||
#include <folly/io/async/EventBase.h>
|
||||
#include <folly/io/async/EventBaseManager.h>
|
||||
|
||||
@@ -17,21 +17,21 @@
|
||||
#pragma once
|
||||
|
||||
#include <folly/futures/Future.h>
|
||||
#include <folly/wangle/channel/ChannelPipeline.h>
|
||||
#include <folly/wangle/channel/Pipeline.h>
|
||||
#include <folly/io/IOBuf.h>
|
||||
#include <folly/io/IOBufQueue.h>
|
||||
|
||||
namespace folly { namespace wangle {
|
||||
|
||||
template <class Rin, class Rout = Rin, class Win = Rout, class Wout = Rin>
|
||||
class ChannelHandler {
|
||||
class Handler {
|
||||
public:
|
||||
typedef Rin rin;
|
||||
typedef Rout rout;
|
||||
typedef Win win;
|
||||
typedef Wout wout;
|
||||
typedef ChannelHandlerContext<Rout, Wout> Context;
|
||||
virtual ~ChannelHandler() {}
|
||||
typedef HandlerContext<Rout, Wout> Context;
|
||||
virtual ~Handler() {}
|
||||
|
||||
virtual void read(Context* ctx, Rin msg) = 0;
|
||||
virtual void readEOF(Context* ctx) {
|
||||
@@ -56,34 +56,34 @@ class ChannelHandler {
|
||||
// Other sorts of things we might want, all shamelessly stolen from Netty
|
||||
// inbound
|
||||
virtual void exceptionCaught(
|
||||
ChannelHandlerContext* ctx,
|
||||
HandlerContext* ctx,
|
||||
exception_wrapper e) {}
|
||||
virtual void channelRegistered(ChannelHandlerContext* ctx) {}
|
||||
virtual void channelUnregistered(ChannelHandlerContext* ctx) {}
|
||||
virtual void channelActive(ChannelHandlerContext* ctx) {}
|
||||
virtual void channelInactive(ChannelHandlerContext* ctx) {}
|
||||
virtual void channelReadComplete(ChannelHandlerContext* ctx) {}
|
||||
virtual void userEventTriggered(ChannelHandlerContext* ctx, void* evt) {}
|
||||
virtual void channelWritabilityChanged(ChannelHandlerContext* ctx) {}
|
||||
virtual void channelRegistered(HandlerContext* ctx) {}
|
||||
virtual void channelUnregistered(HandlerContext* ctx) {}
|
||||
virtual void channelActive(HandlerContext* ctx) {}
|
||||
virtual void channelInactive(HandlerContext* ctx) {}
|
||||
virtual void channelReadComplete(HandlerContext* ctx) {}
|
||||
virtual void userEventTriggered(HandlerContext* ctx, void* evt) {}
|
||||
virtual void channelWritabilityChanged(HandlerContext* ctx) {}
|
||||
|
||||
// outbound
|
||||
virtual Future<void> bind(
|
||||
ChannelHandlerContext* ctx,
|
||||
HandlerContext* ctx,
|
||||
SocketAddress localAddress) {}
|
||||
virtual Future<void> connect(
|
||||
ChannelHandlerContext* ctx,
|
||||
HandlerContext* ctx,
|
||||
SocketAddress remoteAddress, SocketAddress localAddress) {}
|
||||
virtual Future<void> disconnect(ChannelHandlerContext* ctx) {}
|
||||
virtual Future<void> deregister(ChannelHandlerContext* ctx) {}
|
||||
virtual Future<void> read(ChannelHandlerContext* ctx) {}
|
||||
virtual void flush(ChannelHandlerContext* ctx) {}
|
||||
virtual Future<void> disconnect(HandlerContext* ctx) {}
|
||||
virtual Future<void> deregister(HandlerContext* ctx) {}
|
||||
virtual Future<void> read(HandlerContext* ctx) {}
|
||||
virtual void flush(HandlerContext* ctx) {}
|
||||
*/
|
||||
};
|
||||
|
||||
template <class R, class W = R>
|
||||
class ChannelHandlerAdapter : public ChannelHandler<R, R, W, W> {
|
||||
class HandlerAdapter : public Handler<R, R, W, W> {
|
||||
public:
|
||||
typedef typename ChannelHandler<R, R, W, W>::Context Context;
|
||||
typedef typename Handler<R, R, W, W>::Context Context;
|
||||
|
||||
void read(Context* ctx, R msg) override {
|
||||
ctx->fireRead(std::forward<R>(msg));
|
||||
@@ -94,32 +94,32 @@ class ChannelHandlerAdapter : public ChannelHandler<R, R, W, W> {
|
||||
}
|
||||
};
|
||||
|
||||
typedef ChannelHandlerAdapter<IOBufQueue&, std::unique_ptr<IOBuf>>
|
||||
typedef HandlerAdapter<IOBufQueue&, std::unique_ptr<IOBuf>>
|
||||
BytesToBytesHandler;
|
||||
|
||||
template <class Handler, bool Shared = true>
|
||||
class ChannelHandlerPtr : public ChannelHandler<
|
||||
typename Handler::rin,
|
||||
typename Handler::rout,
|
||||
typename Handler::win,
|
||||
typename Handler::wout> {
|
||||
template <class HandlerT, bool Shared = true>
|
||||
class HandlerPtr : public Handler<
|
||||
typename HandlerT::rin,
|
||||
typename HandlerT::rout,
|
||||
typename HandlerT::win,
|
||||
typename HandlerT::wout> {
|
||||
public:
|
||||
typedef typename std::conditional<
|
||||
Shared,
|
||||
std::shared_ptr<Handler>,
|
||||
Handler*>::type
|
||||
HandlerPtr;
|
||||
std::shared_ptr<HandlerT>,
|
||||
HandlerT*>::type
|
||||
Ptr;
|
||||
|
||||
typedef typename Handler::Context Context;
|
||||
typedef typename HandlerT::Context Context;
|
||||
|
||||
explicit ChannelHandlerPtr(HandlerPtr handler)
|
||||
explicit HandlerPtr(Ptr handler)
|
||||
: handler_(std::move(handler)) {}
|
||||
|
||||
HandlerPtr getHandler() {
|
||||
Ptr getHandler() {
|
||||
return handler_;
|
||||
}
|
||||
|
||||
void setHandler(HandlerPtr handler) {
|
||||
void setHandler(Ptr handler) {
|
||||
if (handler == handler_) {
|
||||
return;
|
||||
}
|
||||
@@ -163,9 +163,9 @@ class ChannelHandlerPtr : public ChannelHandler<
|
||||
}
|
||||
}
|
||||
|
||||
void read(Context* ctx, typename Handler::rin msg) override {
|
||||
void read(Context* ctx, typename HandlerT::rin msg) override {
|
||||
DCHECK(handler_);
|
||||
handler_->read(ctx, std::forward<typename Handler::rin>(msg));
|
||||
handler_->read(ctx, std::forward<typename HandlerT::rin>(msg));
|
||||
}
|
||||
|
||||
void readEOF(Context* ctx) override {
|
||||
@@ -178,9 +178,9 @@ class ChannelHandlerPtr : public ChannelHandler<
|
||||
handler_->readException(ctx, std::move(e));
|
||||
}
|
||||
|
||||
Future<void> write(Context* ctx, typename Handler::win msg) override {
|
||||
Future<void> write(Context* ctx, typename HandlerT::win msg) override {
|
||||
DCHECK(handler_);
|
||||
return handler_->write(ctx, std::forward<typename Handler::win>(msg));
|
||||
return handler_->write(ctx, std::forward<typename HandlerT::win>(msg));
|
||||
}
|
||||
|
||||
Future<void> close(Context* ctx) override {
|
||||
@@ -190,7 +190,7 @@ class ChannelHandlerPtr : public ChannelHandler<
|
||||
|
||||
private:
|
||||
Context* ctx_;
|
||||
HandlerPtr handler_;
|
||||
Ptr handler_;
|
||||
};
|
||||
|
||||
}}
|
||||
+16
-16
@@ -23,9 +23,9 @@
|
||||
namespace folly { namespace wangle {
|
||||
|
||||
template <class In, class Out>
|
||||
class ChannelHandlerContext {
|
||||
class HandlerContext {
|
||||
public:
|
||||
virtual ~ChannelHandlerContext() {}
|
||||
virtual ~HandlerContext() {}
|
||||
|
||||
virtual void fireRead(In msg) = 0;
|
||||
virtual void fireReadEOF() = 0;
|
||||
@@ -73,27 +73,27 @@ class PipelineContext {
|
||||
};
|
||||
|
||||
template <class In>
|
||||
class InboundChannelHandlerContext {
|
||||
class InboundHandlerContext {
|
||||
public:
|
||||
virtual ~InboundChannelHandlerContext() {}
|
||||
virtual ~InboundHandlerContext() {}
|
||||
virtual void read(In msg) = 0;
|
||||
virtual void readEOF() = 0;
|
||||
virtual void readException(exception_wrapper e) = 0;
|
||||
};
|
||||
|
||||
template <class Out>
|
||||
class OutboundChannelHandlerContext {
|
||||
class OutboundHandlerContext {
|
||||
public:
|
||||
virtual ~OutboundChannelHandlerContext() {}
|
||||
virtual ~OutboundHandlerContext() {}
|
||||
virtual Future<void> write(Out msg) = 0;
|
||||
virtual Future<void> close() = 0;
|
||||
};
|
||||
|
||||
template <class P, class H>
|
||||
class ContextImpl : public ChannelHandlerContext<typename H::rout,
|
||||
class ContextImpl : public HandlerContext<typename H::rout,
|
||||
typename H::wout>,
|
||||
public InboundChannelHandlerContext<typename H::rin>,
|
||||
public OutboundChannelHandlerContext<typename H::win>,
|
||||
public InboundHandlerContext<typename H::rin>,
|
||||
public OutboundHandlerContext<typename H::win>,
|
||||
public PipelineContext {
|
||||
public:
|
||||
typedef typename H::rin Rin;
|
||||
@@ -118,7 +118,7 @@ class ContextImpl : public ChannelHandlerContext<typename H::rout,
|
||||
|
||||
// PipelineContext overrides
|
||||
void setNextIn(PipelineContext* ctx) override {
|
||||
auto nextIn = dynamic_cast<InboundChannelHandlerContext<Rout>*>(ctx);
|
||||
auto nextIn = dynamic_cast<InboundHandlerContext<Rout>*>(ctx);
|
||||
if (nextIn) {
|
||||
nextIn_ = nextIn;
|
||||
} else {
|
||||
@@ -127,7 +127,7 @@ class ContextImpl : public ChannelHandlerContext<typename H::rout,
|
||||
}
|
||||
|
||||
void setNextOut(PipelineContext* ctx) override {
|
||||
auto nextOut = dynamic_cast<OutboundChannelHandlerContext<Wout>*>(ctx);
|
||||
auto nextOut = dynamic_cast<OutboundHandlerContext<Wout>*>(ctx);
|
||||
if (nextOut) {
|
||||
nextOut_ = nextOut;
|
||||
} else {
|
||||
@@ -145,7 +145,7 @@ class ContextImpl : public ChannelHandlerContext<typename H::rout,
|
||||
handler_.detachTransport(this);
|
||||
}
|
||||
|
||||
// ChannelHandlerContext overrides
|
||||
// HandlerContext overrides
|
||||
void fireRead(Rout msg) override {
|
||||
typename P::DestructorGuard dg(static_cast<DelayedDestruction*>(pipeline_));
|
||||
if (nextIn_) {
|
||||
@@ -215,7 +215,7 @@ class ContextImpl : public ChannelHandlerContext<typename H::rout,
|
||||
return pipeline_->getReadBufferSettings();
|
||||
}
|
||||
|
||||
// InboundChannelHandlerContext overrides
|
||||
// InboundHandlerContext overrides
|
||||
void read(Rin msg) override {
|
||||
typename P::DestructorGuard dg(static_cast<DelayedDestruction*>(pipeline_));
|
||||
handler_.read(this, std::forward<Rin>(msg));
|
||||
@@ -231,7 +231,7 @@ class ContextImpl : public ChannelHandlerContext<typename H::rout,
|
||||
handler_.readException(this, std::move(e));
|
||||
}
|
||||
|
||||
// OutboundChannelHandlerContext overrides
|
||||
// OutboundHandlerContext overrides
|
||||
Future<void> write(Win msg) override {
|
||||
typename P::DestructorGuard dg(static_cast<DelayedDestruction*>(pipeline_));
|
||||
return handler_.write(this, std::forward<Win>(msg));
|
||||
@@ -245,8 +245,8 @@ class ContextImpl : public ChannelHandlerContext<typename H::rout,
|
||||
private:
|
||||
P* pipeline_;
|
||||
H handler_;
|
||||
InboundChannelHandlerContext<Rout>* nextIn_{nullptr};
|
||||
OutboundChannelHandlerContext<Wout>* nextOut_{nullptr};
|
||||
InboundHandlerContext<Rout>* nextIn_{nullptr};
|
||||
OutboundHandlerContext<Wout>* nextOut_{nullptr};
|
||||
};
|
||||
|
||||
}}
|
||||
@@ -16,7 +16,7 @@
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <folly/wangle/channel/ChannelHandler.h>
|
||||
#include <folly/wangle/channel/Handler.h>
|
||||
#include <folly/io/async/EventBase.h>
|
||||
#include <folly/io/async/EventBaseManager.h>
|
||||
#include <folly/io/IOBuf.h>
|
||||
|
||||
@@ -16,7 +16,7 @@
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <folly/wangle/channel/ChannelHandlerContext.h>
|
||||
#include <folly/wangle/channel/HandlerContext.h>
|
||||
#include <folly/futures/Future.h>
|
||||
#include <folly/io/async/AsyncTransport.h>
|
||||
#include <folly/io/async/DelayedDestruction.h>
|
||||
@@ -31,13 +31,13 @@ namespace folly { namespace wangle {
|
||||
* W is the outbound type, i.e. outbound calls start with pipeline.write(W)
|
||||
*/
|
||||
template <class R, class W, class... Handlers>
|
||||
class ChannelPipeline;
|
||||
class Pipeline;
|
||||
|
||||
template <class R, class W>
|
||||
class ChannelPipeline<R, W> : public DelayedDestruction {
|
||||
class Pipeline<R, W> : public DelayedDestruction {
|
||||
public:
|
||||
ChannelPipeline() {}
|
||||
~ChannelPipeline() {}
|
||||
Pipeline() {}
|
||||
~Pipeline() {}
|
||||
|
||||
std::shared_ptr<AsyncTransport> getTransport() {
|
||||
return transport_;
|
||||
@@ -80,17 +80,17 @@ class ChannelPipeline<R, W> : public DelayedDestruction {
|
||||
}
|
||||
|
||||
template <class H>
|
||||
ChannelPipeline& addBack(H&& handler) {
|
||||
ctxs_.push_back(folly::make_unique<ContextImpl<ChannelPipeline, H>>(
|
||||
Pipeline& addBack(H&& handler) {
|
||||
ctxs_.push_back(folly::make_unique<ContextImpl<Pipeline, H>>(
|
||||
this, std::forward<H>(handler)));
|
||||
return *this;
|
||||
}
|
||||
|
||||
template <class H>
|
||||
ChannelPipeline& addFront(H&& handler) {
|
||||
Pipeline& addFront(H&& handler) {
|
||||
ctxs_.insert(
|
||||
ctxs_.begin(),
|
||||
folly::make_unique<ContextImpl<ChannelPipeline, H>>(
|
||||
folly::make_unique<ContextImpl<Pipeline, H>>(
|
||||
this,
|
||||
std::forward<H>(handler)));
|
||||
return *this;
|
||||
@@ -98,15 +98,15 @@ class ChannelPipeline<R, W> : public DelayedDestruction {
|
||||
|
||||
template <class H>
|
||||
H* getHandler(int i) {
|
||||
auto ctx = dynamic_cast<ContextImpl<ChannelPipeline, H>*>(ctxs_[i].get());
|
||||
auto ctx = dynamic_cast<ContextImpl<Pipeline, H>*>(ctxs_[i].get());
|
||||
CHECK(ctx);
|
||||
return ctx->getHandler();
|
||||
}
|
||||
|
||||
void finalize() {
|
||||
finalizeHelper();
|
||||
InboundChannelHandlerContext<R>* front;
|
||||
front_ = dynamic_cast<InboundChannelHandlerContext<R>*>(
|
||||
InboundHandlerContext<R>* front;
|
||||
front_ = dynamic_cast<InboundHandlerContext<R>*>(
|
||||
ctxs_.front().get());
|
||||
if (!front_) {
|
||||
throw std::invalid_argument("wrong type for first handler");
|
||||
@@ -114,7 +114,7 @@ class ChannelPipeline<R, W> : public DelayedDestruction {
|
||||
}
|
||||
|
||||
protected:
|
||||
explicit ChannelPipeline(bool shouldFinalize) {
|
||||
explicit Pipeline(bool shouldFinalize) {
|
||||
CHECK(!shouldFinalize);
|
||||
}
|
||||
|
||||
@@ -127,7 +127,7 @@ class ChannelPipeline<R, W> : public DelayedDestruction {
|
||||
ctxs_[i]->link(ctxs_[i+1].get());
|
||||
}
|
||||
|
||||
back_ = dynamic_cast<OutboundChannelHandlerContext<W>*>(ctxs_.back().get());
|
||||
back_ = dynamic_cast<OutboundHandlerContext<W>*>(ctxs_.back().get());
|
||||
if (!back_) {
|
||||
throw std::invalid_argument("wrong type for last handler");
|
||||
}
|
||||
@@ -154,23 +154,23 @@ class ChannelPipeline<R, W> : public DelayedDestruction {
|
||||
transport_ = nullptr;
|
||||
}
|
||||
|
||||
OutboundChannelHandlerContext<W>* back_{nullptr};
|
||||
OutboundHandlerContext<W>* back_{nullptr};
|
||||
|
||||
private:
|
||||
InboundChannelHandlerContext<R>* front_{nullptr};
|
||||
InboundHandlerContext<R>* front_{nullptr};
|
||||
std::vector<std::unique_ptr<PipelineContext>> ctxs_;
|
||||
};
|
||||
|
||||
template <class R, class W, class Handler, class... Handlers>
|
||||
class ChannelPipeline<R, W, Handler, Handlers...>
|
||||
: public ChannelPipeline<R, W, Handlers...> {
|
||||
class Pipeline<R, W, Handler, Handlers...>
|
||||
: public Pipeline<R, W, Handlers...> {
|
||||
protected:
|
||||
template <class HandlerArg, class... HandlersArgs>
|
||||
ChannelPipeline(
|
||||
Pipeline(
|
||||
bool shouldFinalize,
|
||||
HandlerArg&& handlerArg,
|
||||
HandlersArgs&&... handlersArgs)
|
||||
: ChannelPipeline<R, W, Handlers...>(
|
||||
: Pipeline<R, W, Handlers...>(
|
||||
false,
|
||||
std::forward<HandlersArgs>(handlersArgs)...),
|
||||
ctx_(this, std::forward<HandlerArg>(handlerArg)) {
|
||||
@@ -181,76 +181,76 @@ class ChannelPipeline<R, W, Handler, Handlers...>
|
||||
|
||||
public:
|
||||
template <class... HandlersArgs>
|
||||
explicit ChannelPipeline(HandlersArgs&&... handlersArgs)
|
||||
: ChannelPipeline(true, std::forward<HandlersArgs>(handlersArgs)...) {}
|
||||
explicit Pipeline(HandlersArgs&&... handlersArgs)
|
||||
: Pipeline(true, std::forward<HandlersArgs>(handlersArgs)...) {}
|
||||
|
||||
~ChannelPipeline() {}
|
||||
~Pipeline() {}
|
||||
|
||||
void read(R msg) {
|
||||
typename ChannelPipeline<R, W>::DestructorGuard dg(
|
||||
typename Pipeline<R, W>::DestructorGuard dg(
|
||||
static_cast<DelayedDestruction*>(this));
|
||||
front_->read(std::forward<R>(msg));
|
||||
}
|
||||
|
||||
void readEOF() {
|
||||
typename ChannelPipeline<R, W>::DestructorGuard dg(
|
||||
typename Pipeline<R, W>::DestructorGuard dg(
|
||||
static_cast<DelayedDestruction*>(this));
|
||||
front_->readEOF();
|
||||
}
|
||||
|
||||
void readException(exception_wrapper e) {
|
||||
typename ChannelPipeline<R, W>::DestructorGuard dg(
|
||||
typename Pipeline<R, W>::DestructorGuard dg(
|
||||
static_cast<DelayedDestruction*>(this));
|
||||
front_->readException(std::move(e));
|
||||
}
|
||||
|
||||
Future<void> write(W msg) {
|
||||
typename ChannelPipeline<R, W>::DestructorGuard dg(
|
||||
typename Pipeline<R, W>::DestructorGuard dg(
|
||||
static_cast<DelayedDestruction*>(this));
|
||||
return back_->write(std::forward<W>(msg));
|
||||
}
|
||||
|
||||
Future<void> close() {
|
||||
typename ChannelPipeline<R, W>::DestructorGuard dg(
|
||||
typename Pipeline<R, W>::DestructorGuard dg(
|
||||
static_cast<DelayedDestruction*>(this));
|
||||
return back_->close();
|
||||
}
|
||||
|
||||
void attachTransport(
|
||||
std::shared_ptr<AsyncTransport> transport) {
|
||||
typename ChannelPipeline<R, W>::DestructorGuard dg(
|
||||
typename Pipeline<R, W>::DestructorGuard dg(
|
||||
static_cast<DelayedDestruction*>(this));
|
||||
CHECK((!ChannelPipeline<R, W>::transport_));
|
||||
ChannelPipeline<R, W, Handlers...>::attachTransport(std::move(transport));
|
||||
CHECK((!Pipeline<R, W>::transport_));
|
||||
Pipeline<R, W, Handlers...>::attachTransport(std::move(transport));
|
||||
forEachCtx([&](PipelineContext* ctx){
|
||||
ctx->attachTransport();
|
||||
});
|
||||
}
|
||||
|
||||
void detachTransport() {
|
||||
typename ChannelPipeline<R, W>::DestructorGuard dg(
|
||||
typename Pipeline<R, W>::DestructorGuard dg(
|
||||
static_cast<DelayedDestruction*>(this));
|
||||
ChannelPipeline<R, W, Handlers...>::detachTransport();
|
||||
Pipeline<R, W, Handlers...>::detachTransport();
|
||||
forEachCtx([&](PipelineContext* ctx){
|
||||
ctx->detachTransport();
|
||||
});
|
||||
}
|
||||
|
||||
std::shared_ptr<AsyncTransport> getTransport() {
|
||||
return ChannelPipeline<R, W>::transport_;
|
||||
return Pipeline<R, W>::transport_;
|
||||
}
|
||||
|
||||
template <class H>
|
||||
ChannelPipeline& addBack(H&& handler) {
|
||||
ChannelPipeline<R, W>::addBack(std::move(handler));
|
||||
Pipeline& addBack(H&& handler) {
|
||||
Pipeline<R, W>::addBack(std::move(handler));
|
||||
return *this;
|
||||
}
|
||||
|
||||
template <class H>
|
||||
ChannelPipeline& addFront(H&& handler) {
|
||||
Pipeline& addFront(H&& handler) {
|
||||
ctxs_.insert(
|
||||
ctxs_.begin(),
|
||||
folly::make_unique<ContextImpl<ChannelPipeline, H>>(
|
||||
folly::make_unique<ContextImpl<Pipeline, H>>(
|
||||
this,
|
||||
std::move(handler)));
|
||||
return *this;
|
||||
@@ -259,11 +259,11 @@ class ChannelPipeline<R, W, Handler, Handlers...>
|
||||
template <class H>
|
||||
H* getHandler(size_t i) {
|
||||
if (i > ctxs_.size()) {
|
||||
return ChannelPipeline<R, W, Handlers...>::template getHandler<H>(
|
||||
return Pipeline<R, W, Handlers...>::template getHandler<H>(
|
||||
i - (ctxs_.size() + 1));
|
||||
} else {
|
||||
auto pctx = (i == ctxs_.size()) ? &ctx_ : ctxs_[i].get();
|
||||
auto ctx = dynamic_cast<ContextImpl<ChannelPipeline, H>*>(pctx);
|
||||
auto ctx = dynamic_cast<ContextImpl<Pipeline, H>*>(pctx);
|
||||
return ctx->getHandler();
|
||||
}
|
||||
}
|
||||
@@ -271,7 +271,7 @@ class ChannelPipeline<R, W, Handler, Handlers...>
|
||||
void finalize() {
|
||||
finalizeHelper();
|
||||
auto ctx = ctxs_.empty() ? &ctx_ : ctxs_.front().get();
|
||||
front_ = dynamic_cast<InboundChannelHandlerContext<R>*>(ctx);
|
||||
front_ = dynamic_cast<InboundHandlerContext<R>*>(ctx);
|
||||
if (!front_) {
|
||||
throw std::invalid_argument("wrong type for first handler");
|
||||
}
|
||||
@@ -279,12 +279,12 @@ class ChannelPipeline<R, W, Handler, Handlers...>
|
||||
|
||||
protected:
|
||||
void finalizeHelper() {
|
||||
ChannelPipeline<R, W, Handlers...>::finalizeHelper();
|
||||
back_ = ChannelPipeline<R, W, Handlers...>::back_;
|
||||
Pipeline<R, W, Handlers...>::finalizeHelper();
|
||||
back_ = Pipeline<R, W, Handlers...>::back_;
|
||||
if (!back_) {
|
||||
auto is_at_end = ChannelPipeline<R, W, Handlers...>::is_end;
|
||||
auto is_at_end = Pipeline<R, W, Handlers...>::is_end;
|
||||
CHECK(is_at_end);
|
||||
back_ = dynamic_cast<OutboundChannelHandlerContext<W>*>(&ctx_);
|
||||
back_ = dynamic_cast<OutboundHandlerContext<W>*>(&ctx_);
|
||||
if (!back_) {
|
||||
throw std::invalid_argument("wrong type for last handler");
|
||||
}
|
||||
@@ -297,7 +297,7 @@ class ChannelPipeline<R, W, Handler, Handlers...>
|
||||
ctxs_.back()->link(&ctx_);
|
||||
}
|
||||
|
||||
auto nextFront = ChannelPipeline<R, W, Handlers...>::getLocalFront();
|
||||
auto nextFront = Pipeline<R, W, Handlers...>::getLocalFront();
|
||||
if (nextFront) {
|
||||
ctx_.link(nextFront);
|
||||
}
|
||||
@@ -308,8 +308,8 @@ class ChannelPipeline<R, W, Handler, Handlers...>
|
||||
}
|
||||
|
||||
static const bool is_end{false};
|
||||
InboundChannelHandlerContext<R>* front_{nullptr};
|
||||
OutboundChannelHandlerContext<W>* back_{nullptr};
|
||||
InboundHandlerContext<R>* front_{nullptr};
|
||||
OutboundHandlerContext<W>* back_{nullptr};
|
||||
|
||||
private:
|
||||
template <class F>
|
||||
@@ -320,7 +320,7 @@ class ChannelPipeline<R, W, Handler, Handlers...>
|
||||
func(&ctx_);
|
||||
}
|
||||
|
||||
ContextImpl<ChannelPipeline, Handler> ctx_;
|
||||
ContextImpl<Pipeline, Handler> ctx_;
|
||||
std::vector<std::unique_ptr<PipelineContext>> ctxs_;
|
||||
};
|
||||
|
||||
+6
-6
@@ -16,18 +16,18 @@
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <folly/wangle/channel/ChannelHandler.h>
|
||||
#include <folly/wangle/channel/Handler.h>
|
||||
#include <gmock/gmock.h>
|
||||
|
||||
namespace folly { namespace wangle {
|
||||
|
||||
template <class Rin, class Rout = Rin, class Win = Rout, class Wout = Rin>
|
||||
class MockChannelHandler : public ChannelHandler<Rin, Rout, Win, Wout> {
|
||||
class MockHandler : public Handler<Rin, Rout, Win, Wout> {
|
||||
public:
|
||||
typedef typename ChannelHandler<Rin, Rout, Win, Wout>::Context Context;
|
||||
typedef typename Handler<Rin, Rout, Win, Wout>::Context Context;
|
||||
|
||||
MockChannelHandler() = default;
|
||||
MockChannelHandler(MockChannelHandler&&) = default;
|
||||
MockHandler() = default;
|
||||
MockHandler(MockHandler&&) = default;
|
||||
|
||||
#ifdef __clang__
|
||||
# pragma clang diagnostic push
|
||||
@@ -70,6 +70,6 @@ class MockChannelHandler : public ChannelHandler<Rin, Rout, Win, Wout> {
|
||||
};
|
||||
|
||||
template <class R, class W = R>
|
||||
using MockChannelHandlerAdapter = MockChannelHandler<R, R, W, W>;
|
||||
using MockHandlerAdapter = MockHandler<R, R, W, W>;
|
||||
|
||||
}}
|
||||
@@ -14,9 +14,9 @@
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
#include <folly/wangle/channel/ChannelPipeline.h>
|
||||
#include <folly/wangle/channel/Pipeline.h>
|
||||
#include <folly/wangle/channel/OutputBufferingHandler.h>
|
||||
#include <folly/wangle/channel/test/MockChannelHandler.h>
|
||||
#include <folly/wangle/channel/test/MockHandler.h>
|
||||
#include <folly/io/async/AsyncSocket.h>
|
||||
#include <gmock/gmock.h>
|
||||
#include <gtest/gtest.h>
|
||||
@@ -25,18 +25,18 @@ using namespace folly;
|
||||
using namespace folly::wangle;
|
||||
using namespace testing;
|
||||
|
||||
typedef StrictMock<MockChannelHandlerAdapter<
|
||||
typedef StrictMock<MockHandlerAdapter<
|
||||
IOBufQueue&,
|
||||
std::unique_ptr<IOBuf>>>
|
||||
MockHandler;
|
||||
MockBytesHandler;
|
||||
|
||||
MATCHER_P(IOBufContains, str, "") { return arg->moveToFbString() == str; }
|
||||
|
||||
TEST(OutputBufferingHandlerTest, Basic) {
|
||||
MockHandler mockHandler;
|
||||
MockBytesHandler mockHandler;
|
||||
EXPECT_CALL(mockHandler, attachPipeline(_));
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>,
|
||||
ChannelHandlerPtr<MockHandler, false>,
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>,
|
||||
HandlerPtr<MockBytesHandler, false>,
|
||||
OutputBufferingHandler>
|
||||
pipeline(&mockHandler, OutputBufferingHandler{});
|
||||
|
||||
|
||||
+29
-29
@@ -14,11 +14,11 @@
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
#include <folly/wangle/channel/ChannelHandler.h>
|
||||
#include <folly/wangle/channel/ChannelPipeline.h>
|
||||
#include <folly/wangle/channel/Handler.h>
|
||||
#include <folly/wangle/channel/Pipeline.h>
|
||||
#include <folly/wangle/channel/AsyncSocketHandler.h>
|
||||
#include <folly/wangle/channel/OutputBufferingHandler.h>
|
||||
#include <folly/wangle/channel/test/MockChannelHandler.h>
|
||||
#include <folly/wangle/channel/test/MockHandler.h>
|
||||
#include <gmock/gmock.h>
|
||||
#include <gtest/gtest.h>
|
||||
|
||||
@@ -26,8 +26,8 @@ using namespace folly;
|
||||
using namespace folly::wangle;
|
||||
using namespace testing;
|
||||
|
||||
typedef StrictMock<MockChannelHandlerAdapter<int, int>> IntHandler;
|
||||
typedef ChannelHandlerPtr<IntHandler, false> IntHandlerPtr;
|
||||
typedef StrictMock<MockHandlerAdapter<int, int>> IntHandler;
|
||||
typedef HandlerPtr<IntHandler, false> IntHandlerPtr;
|
||||
|
||||
ACTION(FireRead) {
|
||||
arg0->fireRead(arg1);
|
||||
@@ -50,12 +50,12 @@ ACTION(FireClose) {
|
||||
}
|
||||
|
||||
// Test move only types, among other things
|
||||
TEST(ChannelTest, RealHandlersCompile) {
|
||||
TEST(PipelineTest, RealHandlersCompile) {
|
||||
EventBase eb;
|
||||
auto socket = AsyncSocket::newSocket(&eb);
|
||||
// static
|
||||
{
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>,
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>,
|
||||
AsyncSocketHandler,
|
||||
OutputBufferingHandler>
|
||||
pipeline{AsyncSocketHandler(socket), OutputBufferingHandler()};
|
||||
@@ -64,7 +64,7 @@ TEST(ChannelTest, RealHandlersCompile) {
|
||||
}
|
||||
// dynamic
|
||||
{
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
pipeline
|
||||
.addBack(AsyncSocketHandler(socket))
|
||||
.addBack(OutputBufferingHandler())
|
||||
@@ -75,14 +75,14 @@ TEST(ChannelTest, RealHandlersCompile) {
|
||||
}
|
||||
|
||||
// Test that handlers correctly fire the next handler when directed
|
||||
TEST(ChannelTest, FireActions) {
|
||||
TEST(PipelineTest, FireActions) {
|
||||
IntHandler handler1;
|
||||
IntHandler handler2;
|
||||
|
||||
EXPECT_CALL(handler1, attachPipeline(_));
|
||||
EXPECT_CALL(handler2, attachPipeline(_));
|
||||
|
||||
ChannelPipeline<int, int, IntHandlerPtr, IntHandlerPtr>
|
||||
Pipeline<int, int, IntHandlerPtr, IntHandlerPtr>
|
||||
pipeline(&handler1, &handler2);
|
||||
|
||||
EXPECT_CALL(handler1, read_(_, _)).WillOnce(FireRead());
|
||||
@@ -111,10 +111,10 @@ TEST(ChannelTest, FireActions) {
|
||||
|
||||
// Test that nothing bad happens when actions reach the end of the pipeline
|
||||
// (a warning will be logged, however)
|
||||
TEST(ChannelTest, ReachEndOfPipeline) {
|
||||
TEST(PipelineTest, ReachEndOfPipeline) {
|
||||
IntHandler handler;
|
||||
EXPECT_CALL(handler, attachPipeline(_));
|
||||
ChannelPipeline<int, int, IntHandlerPtr>
|
||||
Pipeline<int, int, IntHandlerPtr>
|
||||
pipeline(&handler);
|
||||
|
||||
EXPECT_CALL(handler, read_(_, _)).WillOnce(FireRead());
|
||||
@@ -136,14 +136,14 @@ TEST(ChannelTest, ReachEndOfPipeline) {
|
||||
}
|
||||
|
||||
// Test having the last read handler turn around and write
|
||||
TEST(ChannelTest, TurnAround) {
|
||||
TEST(PipelineTest, TurnAround) {
|
||||
IntHandler handler1;
|
||||
IntHandler handler2;
|
||||
|
||||
EXPECT_CALL(handler1, attachPipeline(_));
|
||||
EXPECT_CALL(handler2, attachPipeline(_));
|
||||
|
||||
ChannelPipeline<int, int, IntHandlerPtr, IntHandlerPtr>
|
||||
Pipeline<int, int, IntHandlerPtr, IntHandlerPtr>
|
||||
pipeline(&handler1, &handler2);
|
||||
|
||||
EXPECT_CALL(handler1, read_(_, _)).WillOnce(FireRead());
|
||||
@@ -155,10 +155,10 @@ TEST(ChannelTest, TurnAround) {
|
||||
EXPECT_CALL(handler2, detachPipeline(_));
|
||||
}
|
||||
|
||||
TEST(ChannelTest, DynamicFireActions) {
|
||||
TEST(PipelineTest, DynamicFireActions) {
|
||||
IntHandler handler1, handler2, handler3;
|
||||
EXPECT_CALL(handler2, attachPipeline(_));
|
||||
ChannelPipeline<int, int, IntHandlerPtr>
|
||||
Pipeline<int, int, IntHandlerPtr>
|
||||
pipeline(&handler2);
|
||||
|
||||
EXPECT_CALL(handler1, attachPipeline(_));
|
||||
@@ -189,35 +189,35 @@ TEST(ChannelTest, DynamicFireActions) {
|
||||
}
|
||||
|
||||
template <class Rin, class Rout = Rin, class Win = Rout, class Wout = Rin>
|
||||
class ConcreteChannelHandler : public ChannelHandler<Rin, Rout, Win, Wout> {
|
||||
typedef typename ChannelHandler<Rin, Rout, Win, Wout>::Context Context;
|
||||
class ConcreteHandler : public Handler<Rin, Rout, Win, Wout> {
|
||||
typedef typename Handler<Rin, Rout, Win, Wout>::Context Context;
|
||||
public:
|
||||
void read(Context* ctx, Rin msg) {}
|
||||
Future<void> write(Context* ctx, Win msg) { return makeFuture(); }
|
||||
};
|
||||
|
||||
typedef ChannelHandlerAdapter<std::string, std::string> StringHandler;
|
||||
typedef ConcreteChannelHandler<int, std::string> IntToStringHandler;
|
||||
typedef ConcreteChannelHandler<std::string, int> StringToIntHandler;
|
||||
typedef HandlerAdapter<std::string, std::string> StringHandler;
|
||||
typedef ConcreteHandler<int, std::string> IntToStringHandler;
|
||||
typedef ConcreteHandler<std::string, int> StringToIntHandler;
|
||||
|
||||
TEST(ChannelPipeline, DynamicConstruction) {
|
||||
TEST(Pipeline, DynamicConstruction) {
|
||||
{
|
||||
ChannelPipeline<int, int> pipeline;
|
||||
Pipeline<int, int> pipeline;
|
||||
EXPECT_THROW(
|
||||
pipeline
|
||||
.addBack(ChannelHandlerAdapter<std::string, std::string>{})
|
||||
.addBack(HandlerAdapter<std::string, std::string>{})
|
||||
.finalize(), std::invalid_argument);
|
||||
}
|
||||
{
|
||||
ChannelPipeline<int, int> pipeline;
|
||||
Pipeline<int, int> pipeline;
|
||||
EXPECT_THROW(
|
||||
pipeline
|
||||
.addFront(ChannelHandlerAdapter<std::string, std::string>{})
|
||||
.addFront(HandlerAdapter<std::string, std::string>{})
|
||||
.finalize(),
|
||||
std::invalid_argument);
|
||||
}
|
||||
{
|
||||
ChannelPipeline<std::string, std::string, StringHandler, StringHandler>
|
||||
Pipeline<std::string, std::string, StringHandler, StringHandler>
|
||||
pipeline{StringHandler(), StringHandler()};
|
||||
|
||||
// Exercise both addFront and addBack. Final pipeline is
|
||||
@@ -232,10 +232,10 @@ TEST(ChannelPipeline, DynamicConstruction) {
|
||||
}
|
||||
}
|
||||
|
||||
TEST(ChannelPipeline, AttachTransport) {
|
||||
TEST(Pipeline, AttachTransport) {
|
||||
IntHandler handler;
|
||||
EXPECT_CALL(handler, attachPipeline(_));
|
||||
ChannelPipeline<int, int, IntHandlerPtr>
|
||||
Pipeline<int, int, IntHandlerPtr>
|
||||
pipeline(&handler);
|
||||
|
||||
EventBase eb;
|
||||
@@ -15,12 +15,12 @@
|
||||
*/
|
||||
#pragma once
|
||||
|
||||
#include <folly/wangle/channel/ChannelHandler.h>
|
||||
#include <folly/wangle/channel/Handler.h>
|
||||
|
||||
namespace folly { namespace wangle {
|
||||
|
||||
/**
|
||||
* A ChannelHandler which decodes bytes in a stream-like fashion from
|
||||
* A Handler which decodes bytes in a stream-like fashion from
|
||||
* IOBufQueue to a Message type.
|
||||
*
|
||||
* Frame detection
|
||||
|
||||
@@ -55,7 +55,7 @@ class BytesReflector
|
||||
};
|
||||
|
||||
TEST(FixedLengthFrameDecoder, FailWhenLengthFieldEndOffset) {
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
int called = 0;
|
||||
|
||||
pipeline
|
||||
@@ -90,7 +90,7 @@ TEST(FixedLengthFrameDecoder, FailWhenLengthFieldEndOffset) {
|
||||
}
|
||||
|
||||
TEST(LengthFieldFramePipeline, SimpleTest) {
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
int called = 0;
|
||||
|
||||
pipeline
|
||||
@@ -111,7 +111,7 @@ TEST(LengthFieldFramePipeline, SimpleTest) {
|
||||
}
|
||||
|
||||
TEST(LengthFieldFramePipeline, LittleEndian) {
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
int called = 0;
|
||||
|
||||
pipeline
|
||||
@@ -132,7 +132,7 @@ TEST(LengthFieldFramePipeline, LittleEndian) {
|
||||
}
|
||||
|
||||
TEST(LengthFieldFrameDecoder, Simple) {
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
int called = 0;
|
||||
|
||||
pipeline
|
||||
@@ -163,7 +163,7 @@ TEST(LengthFieldFrameDecoder, Simple) {
|
||||
}
|
||||
|
||||
TEST(LengthFieldFrameDecoder, NoStrip) {
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
int called = 0;
|
||||
|
||||
pipeline
|
||||
@@ -194,7 +194,7 @@ TEST(LengthFieldFrameDecoder, NoStrip) {
|
||||
}
|
||||
|
||||
TEST(LengthFieldFrameDecoder, Adjustment) {
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
int called = 0;
|
||||
|
||||
pipeline
|
||||
@@ -225,7 +225,7 @@ TEST(LengthFieldFrameDecoder, Adjustment) {
|
||||
}
|
||||
|
||||
TEST(LengthFieldFrameDecoder, PreHeader) {
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
int called = 0;
|
||||
|
||||
pipeline
|
||||
@@ -257,7 +257,7 @@ TEST(LengthFieldFrameDecoder, PreHeader) {
|
||||
}
|
||||
|
||||
TEST(LengthFieldFrameDecoder, PostHeader) {
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
int called = 0;
|
||||
|
||||
pipeline
|
||||
@@ -289,7 +289,7 @@ TEST(LengthFieldFrameDecoder, PostHeader) {
|
||||
}
|
||||
|
||||
TEST(LengthFieldFrameDecoderStrip, PrePostHeader) {
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
int called = 0;
|
||||
|
||||
pipeline
|
||||
@@ -322,7 +322,7 @@ TEST(LengthFieldFrameDecoderStrip, PrePostHeader) {
|
||||
}
|
||||
|
||||
TEST(LengthFieldFrameDecoder, StripPrePostHeaderFrameInclHeader) {
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
int called = 0;
|
||||
|
||||
pipeline
|
||||
@@ -355,7 +355,7 @@ TEST(LengthFieldFrameDecoder, StripPrePostHeaderFrameInclHeader) {
|
||||
}
|
||||
|
||||
TEST(LengthFieldFrameDecoder, FailTestLengthFieldEndOffset) {
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
int called = 0;
|
||||
|
||||
pipeline
|
||||
@@ -380,7 +380,7 @@ TEST(LengthFieldFrameDecoder, FailTestLengthFieldEndOffset) {
|
||||
}
|
||||
|
||||
TEST(LengthFieldFrameDecoder, FailTestLengthFieldFrameSize) {
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
int called = 0;
|
||||
|
||||
pipeline
|
||||
@@ -407,7 +407,7 @@ TEST(LengthFieldFrameDecoder, FailTestLengthFieldFrameSize) {
|
||||
}
|
||||
|
||||
TEST(LengthFieldFrameDecoder, FailTestLengthFieldInitialBytes) {
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
int called = 0;
|
||||
|
||||
pipeline
|
||||
@@ -434,7 +434,7 @@ TEST(LengthFieldFrameDecoder, FailTestLengthFieldInitialBytes) {
|
||||
}
|
||||
|
||||
TEST(LineBasedFrameDecoder, Simple) {
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
int called = 0;
|
||||
|
||||
pipeline
|
||||
@@ -485,7 +485,7 @@ TEST(LineBasedFrameDecoder, Simple) {
|
||||
}
|
||||
|
||||
TEST(LineBasedFrameDecoder, SaveDelimiter) {
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
int called = 0;
|
||||
|
||||
pipeline
|
||||
@@ -534,7 +534,7 @@ TEST(LineBasedFrameDecoder, SaveDelimiter) {
|
||||
}
|
||||
|
||||
TEST(LineBasedFrameDecoder, Fail) {
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
int called = 0;
|
||||
|
||||
pipeline
|
||||
@@ -582,7 +582,7 @@ TEST(LineBasedFrameDecoder, Fail) {
|
||||
}
|
||||
|
||||
TEST(LineBasedFrameDecoder, NewLineOnly) {
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
int called = 0;
|
||||
|
||||
pipeline
|
||||
@@ -609,7 +609,7 @@ TEST(LineBasedFrameDecoder, NewLineOnly) {
|
||||
}
|
||||
|
||||
TEST(LineBasedFrameDecoder, CarriageNewLineOnly) {
|
||||
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
Pipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
|
||||
int called = 0;
|
||||
|
||||
pipeline
|
||||
|
||||
@@ -16,17 +16,17 @@
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <folly/wangle/channel/ChannelHandler.h>
|
||||
#include <folly/wangle/channel/Handler.h>
|
||||
|
||||
namespace folly { namespace wangle {
|
||||
|
||||
/*
|
||||
* StringCodec converts a pipeline from IOBufs to std::strings.
|
||||
*/
|
||||
class StringCodec : public ChannelHandler<IOBufQueue&, std::string,
|
||||
std::string, std::unique_ptr<IOBuf>> {
|
||||
class StringCodec : public Handler<IOBufQueue&, std::string,
|
||||
std::string, std::unique_ptr<IOBuf>> {
|
||||
public:
|
||||
typedef typename ChannelHandler<
|
||||
typedef typename Handler<
|
||||
IOBufQueue&, std::string,
|
||||
std::string, std::unique_ptr<IOBuf>>::Context Context;
|
||||
|
||||
|
||||
@@ -15,7 +15,7 @@
|
||||
*/
|
||||
#pragma once
|
||||
|
||||
#include <folly/wangle/channel/ChannelHandler.h>
|
||||
#include <folly/wangle/channel/Handler.h>
|
||||
#include <folly/wangle/service/Service.h>
|
||||
|
||||
namespace folly { namespace wangle {
|
||||
@@ -26,16 +26,16 @@ namespace folly { namespace wangle {
|
||||
* only one request is allowed at a time.
|
||||
*/
|
||||
template <typename Pipeline, typename Req, typename Resp = Req>
|
||||
class SerialClientDispatcher : public ChannelHandlerAdapter<Req, Resp>
|
||||
class SerialClientDispatcher : public HandlerAdapter<Req, Resp>
|
||||
, public Service<Req, Resp> {
|
||||
public:
|
||||
|
||||
typedef typename ChannelHandlerAdapter<Req, Resp>::Context Context;
|
||||
typedef typename HandlerAdapter<Req, Resp>::Context Context;
|
||||
|
||||
void setPipeline(Pipeline* pipeline) {
|
||||
pipeline_ = pipeline;
|
||||
pipeline->addBack(
|
||||
ChannelHandlerPtr<SerialClientDispatcher<Pipeline, Req, Resp>, false>(
|
||||
HandlerPtr<SerialClientDispatcher<Pipeline, Req, Resp>, false>(
|
||||
this));
|
||||
pipeline->finalize();
|
||||
}
|
||||
|
||||
@@ -15,7 +15,7 @@
|
||||
*/
|
||||
#pragma once
|
||||
|
||||
#include <folly/wangle/channel/ChannelHandler.h>
|
||||
#include <folly/wangle/channel/Handler.h>
|
||||
#include <folly/wangle/service/Service.h>
|
||||
|
||||
namespace folly { namespace wangle {
|
||||
@@ -25,10 +25,10 @@ namespace folly { namespace wangle {
|
||||
* Concurrent requests are queued in the pipeline.
|
||||
*/
|
||||
template <typename Req, typename Resp = Req>
|
||||
class SerialServerDispatcher : public ChannelHandlerAdapter<Req, Resp> {
|
||||
class SerialServerDispatcher : public HandlerAdapter<Req, Resp> {
|
||||
public:
|
||||
|
||||
typedef typename ChannelHandlerAdapter<Req, Resp>::Context Context;
|
||||
typedef typename HandlerAdapter<Req, Resp>::Context Context;
|
||||
|
||||
explicit SerialServerDispatcher(Service<Req, Resp>* service)
|
||||
: service_(service) {}
|
||||
|
||||
@@ -20,7 +20,7 @@
|
||||
|
||||
#include <folly/wangle/bootstrap/ServerBootstrap.h>
|
||||
#include <folly/wangle/bootstrap/ClientBootstrap.h>
|
||||
#include <folly/wangle/channel/ChannelPipeline.h>
|
||||
#include <folly/wangle/channel/Pipeline.h>
|
||||
#include <folly/wangle/channel/AsyncSocketHandler.h>
|
||||
|
||||
namespace folly {
|
||||
|
||||
@@ -24,7 +24,7 @@ namespace folly {
|
||||
|
||||
using namespace wangle;
|
||||
|
||||
typedef ChannelPipeline<IOBufQueue&, std::string> Pipeline;
|
||||
typedef Pipeline<IOBufQueue&, std::string> ServicePipeline;
|
||||
|
||||
class EchoService : public Service<std::string, std::string> {
|
||||
public:
|
||||
@@ -42,12 +42,12 @@ class EchoIntService : public Service<std::string, int> {
|
||||
|
||||
template <typename Req, typename Resp>
|
||||
class ServerPipelineFactory
|
||||
: public PipelineFactory<Pipeline> {
|
||||
: public PipelineFactory<ServicePipeline> {
|
||||
public:
|
||||
|
||||
Pipeline* newPipeline(
|
||||
ServicePipeline* newPipeline(
|
||||
std::shared_ptr<AsyncSocket> socket) override {
|
||||
auto pipeline = new Pipeline();
|
||||
auto pipeline = new ServicePipeline();
|
||||
pipeline->addBack(AsyncSocketHandler(socket));
|
||||
pipeline->addBack(StringCodec());
|
||||
pipeline->addBack(SerialServerDispatcher<Req, Resp>(&service_));
|
||||
@@ -61,12 +61,12 @@ class ServerPipelineFactory
|
||||
};
|
||||
|
||||
template <typename Req, typename Resp>
|
||||
class ClientPipelineFactory : public PipelineFactory<Pipeline> {
|
||||
class ClientPipelineFactory : public PipelineFactory<ServicePipeline> {
|
||||
public:
|
||||
|
||||
Pipeline* newPipeline(
|
||||
ServicePipeline* newPipeline(
|
||||
std::shared_ptr<AsyncSocket> socket) override {
|
||||
auto pipeline = new Pipeline();
|
||||
auto pipeline = new ServicePipeline();
|
||||
pipeline->addBack(AsyncSocketHandler(socket));
|
||||
pipeline->addBack(StringCodec());
|
||||
pipeline->template getHandler<AsyncSocketHandler>(0)->attachReadCallback();
|
||||
@@ -101,14 +101,14 @@ TEST(Wangle, ClientServerTest) {
|
||||
int port = 1234;
|
||||
// server
|
||||
|
||||
ServerBootstrap<Pipeline> server;
|
||||
ServerBootstrap<ServicePipeline> server;
|
||||
server.childPipeline(
|
||||
std::make_shared<ServerPipelineFactory<std::string, std::string>>());
|
||||
server.bind(port);
|
||||
|
||||
// client
|
||||
auto client = std::make_shared<ClientBootstrap<Pipeline>>();
|
||||
ClientServiceFactory<Pipeline, std::string, std::string> serviceFactory;
|
||||
auto client = std::make_shared<ClientBootstrap<ServicePipeline>>();
|
||||
ClientServiceFactory<ServicePipeline, std::string, std::string> serviceFactory;
|
||||
client->pipelineFactory(
|
||||
std::make_shared<ClientPipelineFactory<std::string, std::string>>());
|
||||
SocketAddress addr("127.0.0.1", port);
|
||||
|
||||
Reference in New Issue
Block a user