ConnectionAutomaton& connection,

ReactiveSocketMode mode)

: connection_(connection),

nextStreamId_(

mode == ReactiveSocketMode::CLIENT

? 1 /*Streams initiated by a client MUST use

: 2 /*streams initiated by the server MUST use

std::shared_ptr<Subscriber<Payload>> responseSink,

folly::Executor& executor) {

ChannelRequester::Parameters params = {

{connection_.shared_from_this(), getNextStreamId()}, executor};

auto automaton = std::make_shared<ChannelRequester>(params);

connection_.addStream(params.streamId, automaton);

automaton->subscribe(std::move(responseSink));

return automaton;

Payload request,

std::shared_ptr<Subscriber<Payload>> responseSink,

folly::Executor& executor) {

StreamRequester::Parameters params = {

{connection_.shared_from_this(), getNextStreamId()}, executor};

auto automaton =

std::make_shared<StreamRequester>(params, std::move(request));

connection_.addStream(params.streamId, automaton);

automaton->subscribe(std::move(responseSink));

Payload request,

std::shared_ptr<Subscriber<Payload>> responseSink,

folly::Executor& executor) {

SubscriptionRequester::Parameters params = {

{connection_.shared_from_this(), getNextStreamId()}, executor};

auto automaton =

std::make_shared<SubscriptionRequester>(params, std::move(request));

connection_.addStream(params.streamId, automaton);

automaton->subscribe(std::move(responseSink));

Payload payload,

std::shared_ptr<Subscriber<Payload>> responseSink,

folly::Executor& executor) {

RequestResponseRequester::Parameters params = {

{connection_.shared_from_this(), getNextStreamId()}, executor};

auto automaton =

std::make_shared<RequestResponseRequester>(params, std::move(payload));

connection_.addStream(params.streamId, automaton);

automaton->subscribe(std::move(responseSink));

StreamId streamId = nextStreamId_;

nextStreamId_ += 2;

return streamId;

DCHECK(streamId != 0);

if (nextStreamId_ % 2 == streamId % 2) {

// if this is an unknown stream to the socket and this socket is

// generating

// such stream ids, it is an incoming frame on the stream which no longer

// exist

return false;

}

if (streamId <= lastPeerStreamId_) {

// receiving frame for a stream which no longer exists

return false;

}

lastPeerStreamId_ = streamId;

return true;

uint32_t initialRequestN,

StreamId streamId,

folly::Executor& executor) {

ChannelResponder::Parameters params = {

{connection_.shared_from_this(), streamId}, executor};

auto automaton = std::make_shared<ChannelResponder>(initialRequestN, params);

connection_.addStream(streamId, automaton);

return automaton;

uint32_t initialRequestN,

StreamId streamId,

folly::Executor& executor) {

StreamResponder::Parameters params = {

{connection_.shared_from_this(), streamId}, executor};

auto automaton = std::make_shared<StreamResponder>(initialRequestN, params);

connection_.addStream(streamId, automaton);

return automaton;

uint32_t initialRequestN,

StreamId streamId,

folly::Executor& executor) {

SubscriptionResponder::Parameters params = {

{connection_.shared_from_this(), streamId}, executor};

auto automaton =

std::make_shared<SubscriptionResponder>(initialRequestN, params);

connection_.addStream(streamId, automaton);

return automaton;

StreamId streamId,

folly::Executor& executor) {

RequestResponseResponder::Parameters params = {

{connection_.shared_from_this(), streamId}, executor};

auto automaton = std::make_shared<RequestResponseResponder>(params);

connection_.addStream(streamId, automaton);

return automaton;