Commit 2cfb9ed6 authored by James Sedgwick's avatar James Sedgwick Committed by Dave Watson

ChannelPipeline tests and fixes

Summary: As above. This paid off with a couple bugfixes.

Test Plan: run em all

Reviewed By: hans@fb.com

Subscribers: fugalh, njormrod, folly-diffs@

FB internal diff: D1695106

Signature: t1:1695106:1416522038:0c3345aadf954bf346d35b99877e7f8dfcf3ceff
parent 96e30d31
......@@ -123,7 +123,7 @@ class AsyncSocketHandler
void readErr(const AsyncSocketException& ex)
noexcept override {
ctx_->fireReadException(ex);
ctx_->fireReadException(make_exception_wrapper<AsyncSocketException>(ex));
}
private:
......
......@@ -26,6 +26,10 @@
namespace folly { namespace wangle {
/*
* R is the inbound type, i.e. inbound calls start with pipeline.read(R)
* W is the outbound type, i.e. outbound calls start with pipeline.write(W)
*/
template <class R, class W, class... Handlers>
class ChannelPipeline;
......@@ -84,8 +88,11 @@ class ChannelPipeline<R, W> : public DelayedDestruction {
template <class H>
ChannelPipeline& addFront(H&& handler) {
ctxs_.insert(0, folly::make_unique<ContextImpl<ChannelPipeline, H>>(
this, std::forward<H>(handler)));
ctxs_.insert(
ctxs_.begin(),
folly::make_unique<ContextImpl<ChannelPipeline, H>>(
this,
std::forward<H>(handler)));
return *this;
}
......@@ -171,6 +178,7 @@ class ChannelPipeline<R, W, Handler, Handlers...>
finalize();
}
}
public:
template <class... HandlersArgs>
explicit ChannelPipeline(HandlersArgs&&... handlersArgs)
......@@ -195,7 +203,7 @@ class ChannelPipeline<R, W, Handler, Handlers...>
void readException(exception_wrapper e) {
typename ChannelPipeline<R, W>::DestructorGuard dg(
static_cast<DelayedDestruction*>(this));
front_->readEOF(std::move(e));
front_->readException(std::move(e));
}
Future<void> write(W msg) {
......@@ -242,8 +250,11 @@ class ChannelPipeline<R, W, Handler, Handlers...>
template <class H>
ChannelPipeline& addFront(H&& handler) {
ctxs_.insert(0, folly::make_unique<ContextImpl<ChannelPipeline, H>>(
this, std::move(handler)));
ctxs_.insert(
ctxs_.begin(),
folly::make_unique<ContextImpl<ChannelPipeline, H>>(
this,
std::move(handler)));
return *this;
}
......
......@@ -18,6 +18,7 @@
#include <folly/experimental/wangle/channel/ChannelPipeline.h>
#include <folly/experimental/wangle/channel/AsyncSocketHandler.h>
#include <folly/experimental/wangle/channel/OutputBufferingHandler.h>
#include <folly/experimental/wangle/channel/test/MockChannelHandler.h>
#include <folly/io/IOBufQueue.h>
#include <folly/Memory.h>
#include <folly/Conv.h>
......@@ -106,7 +107,7 @@ TEST(ChannelTest, PlzCompile2) {
pipeline.read(42);
}
TEST(ChannelTest, HandlersCompile) {
TEST(ChannelTest, RealHandlersCompile) {
EventBase eb;
auto socket = AsyncSocket::newSocket(&eb);
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
......@@ -115,3 +116,31 @@ TEST(ChannelTest, HandlersCompile) {
.addBack(OutputBufferingHandler())
.finalize();
}
TEST(ChannelTest, MoveOnlyTypesCompile) {
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>,
BytesToBytesHandler,
BytesToBytesHandler>
pipeline(BytesToBytesHandler{}, BytesToBytesHandler{});
pipeline
.addFront(BytesToBytesHandler{})
.addBack(BytesToBytesHandler{})
.finalize();
}
typedef StrictMock<MockChannelHandlerAdapter<int, int>> IntHandler;
ACTION(FireRead) {
arg0->fireRead(arg1);
}
TEST(ChannelTest, Handoffs) {
IntHandler handler1;
IntHandler handler2;
MockChannelHandlerAdapter<int, int> handler2;
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>,
ChannelHandlerPtr<IntHandler, false>,
ChannelHandlerPtr<IntHandler, false>>
pipeline(IntHandler{}, IntHandler{});
pipeline.read(1);
}
/*
* Copyright 2014 Facebook, Inc.
*
* Licensed 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 <folly/experimental/wangle/channel/ChannelHandler.h>
#include <folly/experimental/wangle/channel/ChannelPipeline.h>
#include <folly/experimental/wangle/channel/AsyncSocketHandler.h>
#include <folly/experimental/wangle/channel/OutputBufferingHandler.h>
#include <folly/experimental/wangle/channel/test/MockChannelHandler.h>
#include <gmock/gmock.h>
#include <gtest/gtest.h>
using namespace folly;
using namespace folly::wangle;
using namespace testing;
typedef StrictMock<MockChannelHandlerAdapter<int, int>> IntHandler;
typedef ChannelHandlerPtr<IntHandler, false> IntHandlerPtr;
ACTION(FireRead) {
arg0->fireRead(arg1);
}
ACTION(FireReadEOF) {
arg0->fireReadEOF();
}
ACTION(FireReadException) {
arg0->fireReadException(arg1);
}
ACTION(FireWrite) {
arg0->fireWrite(arg1);
}
ACTION(FireClose) {
arg0->fireClose();
}
// Test move only types, among other things
TEST(ChannelTest, RealHandlersCompile) {
EventBase eb;
auto socket = AsyncSocket::newSocket(&eb);
// static
{
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>,
AsyncSocketHandler,
OutputBufferingHandler>
pipeline{AsyncSocketHandler(socket), OutputBufferingHandler()};
EXPECT_TRUE(pipeline.getHandler<AsyncSocketHandler>(0));
EXPECT_TRUE(pipeline.getHandler<OutputBufferingHandler>(1));
}
// dynamic
{
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>> pipeline;
pipeline
.addBack(AsyncSocketHandler(socket))
.addBack(OutputBufferingHandler())
.finalize();
EXPECT_TRUE(pipeline.getHandler<AsyncSocketHandler>(0));
EXPECT_TRUE(pipeline.getHandler<OutputBufferingHandler>(1));
}
}
// Test that handlers correctly fire the next handler when directed
TEST(ChannelTest, FireActions) {
IntHandler handler1;
IntHandler handler2;
EXPECT_CALL(handler1, attachPipeline(_));
EXPECT_CALL(handler2, attachPipeline(_));
ChannelPipeline<int, int, IntHandlerPtr, IntHandlerPtr>
pipeline(&handler1, &handler2);
EXPECT_CALL(handler1, read_(_, _)).WillOnce(FireRead());
EXPECT_CALL(handler2, read_(_, _)).Times(1);
pipeline.read(1);
EXPECT_CALL(handler1, readEOF(_)).WillOnce(FireReadEOF());
EXPECT_CALL(handler2, readEOF(_)).Times(1);
pipeline.readEOF();
EXPECT_CALL(handler1, readException(_, _)).WillOnce(FireReadException());
EXPECT_CALL(handler2, readException(_, _)).Times(1);
pipeline.readException(make_exception_wrapper<std::runtime_error>("blah"));
EXPECT_CALL(handler2, write_(_, _)).WillOnce(FireWrite());
EXPECT_CALL(handler1, write_(_, _)).Times(1);
EXPECT_NO_THROW(pipeline.write(1).value());
EXPECT_CALL(handler2, close_(_)).WillOnce(FireClose());
EXPECT_CALL(handler1, close_(_)).Times(1);
EXPECT_NO_THROW(pipeline.close().value());
EXPECT_CALL(handler1, detachPipeline(_));
EXPECT_CALL(handler2, detachPipeline(_));
}
// Test that nothing bad happens when actions reach the end of the pipeline
// (a warning will be logged, however)
TEST(ChannelTest, ReachEndOfPipeline) {
IntHandler handler;
EXPECT_CALL(handler, attachPipeline(_));
ChannelPipeline<int, int, IntHandlerPtr>
pipeline(&handler);
EXPECT_CALL(handler, read_(_, _)).WillOnce(FireRead());
pipeline.read(1);
EXPECT_CALL(handler, readEOF(_)).WillOnce(FireReadEOF());
pipeline.readEOF();
EXPECT_CALL(handler, readException(_, _)).WillOnce(FireReadException());
pipeline.readException(make_exception_wrapper<std::runtime_error>("blah"));
EXPECT_CALL(handler, write_(_, _)).WillOnce(FireWrite());
EXPECT_NO_THROW(pipeline.write(1).value());
EXPECT_CALL(handler, close_(_)).WillOnce(FireClose());
EXPECT_NO_THROW(pipeline.close().value());
EXPECT_CALL(handler, detachPipeline(_));
}
// Test having the last read handler turn around and write
TEST(ChannelTest, TurnAround) {
IntHandler handler1;
IntHandler handler2;
EXPECT_CALL(handler1, attachPipeline(_));
EXPECT_CALL(handler2, attachPipeline(_));
ChannelPipeline<int, int, IntHandlerPtr, IntHandlerPtr>
pipeline(&handler1, &handler2);
EXPECT_CALL(handler1, read_(_, _)).WillOnce(FireRead());
EXPECT_CALL(handler2, read_(_, _)).WillOnce(FireWrite());
EXPECT_CALL(handler1, write_(_, _)).Times(1);
pipeline.read(1);
EXPECT_CALL(handler1, detachPipeline(_));
EXPECT_CALL(handler2, detachPipeline(_));
}
TEST(ChannelTest, DynamicFireActions) {
IntHandler handler1, handler2, handler3;
EXPECT_CALL(handler2, attachPipeline(_));
ChannelPipeline<int, int, IntHandlerPtr>
pipeline(&handler2);
EXPECT_CALL(handler1, attachPipeline(_));
EXPECT_CALL(handler3, attachPipeline(_));
pipeline
.addFront(IntHandlerPtr(&handler1))
.addBack(IntHandlerPtr(&handler3))
.finalize();
EXPECT_TRUE(pipeline.getHandler<IntHandlerPtr>(0));
EXPECT_TRUE(pipeline.getHandler<IntHandlerPtr>(1));
EXPECT_TRUE(pipeline.getHandler<IntHandlerPtr>(2));
EXPECT_CALL(handler1, read_(_, _)).WillOnce(FireRead());
EXPECT_CALL(handler2, read_(_, _)).WillOnce(FireRead());
EXPECT_CALL(handler3, read_(_, _)).Times(1);
pipeline.read(1);
EXPECT_CALL(handler3, write_(_, _)).WillOnce(FireWrite());
EXPECT_CALL(handler2, write_(_, _)).WillOnce(FireWrite());
EXPECT_CALL(handler1, write_(_, _)).Times(1);
EXPECT_NO_THROW(pipeline.write(1).value());
EXPECT_CALL(handler1, detachPipeline(_));
EXPECT_CALL(handler2, detachPipeline(_));
EXPECT_CALL(handler3, detachPipeline(_));
}
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;
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;
TEST(ChannelPipeline, DynamicConstruction) {
{
ChannelPipeline<int, int> pipeline;
EXPECT_THROW(
pipeline
.addBack(ChannelHandlerAdapter<std::string, std::string>{})
.finalize(), std::invalid_argument);
}
{
ChannelPipeline<int, int> pipeline;
EXPECT_THROW(
pipeline
.addFront(ChannelHandlerAdapter<std::string, std::string>{})
.finalize(),
std::invalid_argument);
}
{
ChannelPipeline<std::string, std::string, StringHandler, StringHandler>
pipeline{StringHandler(), StringHandler()};
// Exercise both addFront and addBack. Final pipeline is
// StI <-> ItS <-> StS <-> StS <-> StI <-> ItS
EXPECT_NO_THROW(
pipeline
.addFront(IntToStringHandler{})
.addFront(StringToIntHandler{})
.addBack(StringToIntHandler{})
.addBack(IntToStringHandler{})
.finalize());
}
}
TEST(ChannelPipeline, AttachTransport) {
IntHandler handler;
EXPECT_CALL(handler, attachPipeline(_));
ChannelPipeline<int, int, IntHandlerPtr>
pipeline(&handler);
EventBase eb;
auto socket = AsyncSocket::newSocket(&eb);
EXPECT_CALL(handler, attachTransport(_));
pipeline.attachTransport(socket);
EXPECT_CALL(handler, detachTransport(_));
pipeline.detachTransport();
EXPECT_CALL(handler, detachPipeline(_));
}
/*
* Copyright 2014 Facebook, Inc.
*
* Licensed 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.
*/
#pragma once
#include <folly/experimental/wangle/channel/ChannelHandler.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> {
public:
typedef typename ChannelHandler<Rin, Rout, Win, Wout>::Context Context;
MockChannelHandler() = default;
MockChannelHandler(MockChannelHandler&&) = default;
MOCK_METHOD2_T(read_, void(Context*, Rin&));
MOCK_METHOD1_T(readEOF, void(Context*));
MOCK_METHOD2_T(readException, void(Context*, exception_wrapper));
MOCK_METHOD2_T(write_, void(Context*, Win&));
MOCK_METHOD1_T(close_, void(Context*));
MOCK_METHOD1_T(attachPipeline, void(Context*));
MOCK_METHOD1_T(attachTransport, void(Context*));
MOCK_METHOD1_T(detachPipeline, void(Context*));
MOCK_METHOD1_T(detachTransport, void(Context*));
void read(Context* ctx, Rin msg) {
read_(ctx, msg);
}
Future<void> write(Context* ctx, Win msg) override {
return makeFutureTry([&](){
write_(ctx, msg);
});
}
Future<void> close(Context* ctx) override {
return makeFutureTry([&](){
close_(ctx);
});
}
};
template <class R, class W = R>
using MockChannelHandlerAdapter = MockChannelHandler<R, R, W, W>;
}}
......@@ -16,6 +16,7 @@
#include <folly/experimental/wangle/channel/ChannelPipeline.h>
#include <folly/experimental/wangle/channel/OutputBufferingHandler.h>
#include <folly/experimental/wangle/channel/test/MockChannelHandler.h>
#include <folly/io/async/AsyncSocket.h>
#include <gmock/gmock.h>
#include <gtest/gtest.h>
......@@ -24,27 +25,16 @@ using namespace folly;
using namespace folly::wangle;
using namespace testing;
// TODO(jsedgwick) Extract this to somewhere common and fill it out.
template <class R, class W = R>
class MockChannelHandler : public ChannelHandlerAdapter<R, W> {
public:
typedef typename ChannelHandlerAdapter<R, W>::Context Context;
MOCK_METHOD2_T(read, void(Context*, R));
MOCK_METHOD2_T(write_, void(Context*, W&));
Future<void> write(Context* ctx, W msg) override {
write_(ctx, msg);
return makeFuture();
}
};
typedef StrictMock<MockChannelHandler<IOBufQueue&, std::unique_ptr<IOBuf>>>
typedef StrictMock<MockChannelHandlerAdapter<
IOBufQueue&,
std::unique_ptr<IOBuf>>>
MockHandler;
MATCHER_P(IOBufContains, str, "") { return arg->moveToFbString() == str; }
TEST(OutputBufferingHandlerTest, Basic) {
MockHandler mockHandler;
EXPECT_CALL(mockHandler, attachPipeline(_));
ChannelPipeline<IOBufQueue&, std::unique_ptr<IOBuf>,
ChannelHandlerPtr<MockHandler, false>,
OutputBufferingHandler>
......@@ -52,6 +42,7 @@ TEST(OutputBufferingHandlerTest, Basic) {
EventBase eb;
auto socket = AsyncSocket::newSocket(&eb);
EXPECT_CALL(mockHandler, attachTransport(_));
pipeline.attachTransport(socket);
// Buffering should prevent writes until the EB loops, and the writes should
......@@ -64,4 +55,5 @@ TEST(OutputBufferingHandlerTest, Basic) {
eb.loopOnce();
EXPECT_TRUE(f1.isReady());
EXPECT_TRUE(f2.isReady());
EXPECT_CALL(mockHandler, detachPipeline(_));
}
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