rsocket / rsocket-cpp Goto Github PK

View Code? Open in Web Editor NEW

253.0 36.0 99.0 7.31 MB

C++ implementation of RSocket

Home Page: http://rsocket.io

License: Apache License 2.0

CMake 6.66% C++ 72.20% Shell 0.79% C 0.36% Python 20.00%

rsocket-cpp's Introduction

rsocket-cpp

C++ implementation of RSocket

Dependencies

Install folly:

brew install --HEAD folly

Building and running tests

After installing dependencies as above, you can build and run tests with:

# inside root ./rsocket-cpp
mkdir -p build
cd build
cmake -DCMAKE_BUILD_TYPE=DEBUG ../
make -j
./tests

License

By contributing to rsocket-cpp, you agree that your contributions will be licensed under the LICENSE file in the root directory of this source tree.

rsocket-cpp's People

Contributors

Stargazers

Watchers

Forkers

gitter-badger alexstaytuned jrahman-zz cristinag lexs foghina orvid grzmiel th0br0 somasun fbandersnatch tmontgomery ragansa mwinstanley benjchristensen nickgg lehecka jacobkahn junaidkhalid dymk qizhou fleupold lukepg priteshrnandgaonkar stevegury netifi pkrmf oliveiradan yns88 spencerx arushidogra biswapanda simpkins mingtaoy hanbim520 zengyuxing007 weizai118 jknoxville mostroverkhov wez wanener supreme-core wyw0907 pedroalvarado kimsama vdedyukhin chenkangcrack jack-kevin phoad cool-colo linux-china dendisuhubdy forkkit blockspacer welwit gangdao2013 yazici longjohncoder panlaoda benguang feiyunwill xuning97 5l1v3r1 kkathirvel ardieb lincolnphu natyamatsya rayowang groszewn boli-bohan wiltonlazary lujiwei massshock57 rostamn739 wby002 wby-test bruto1 mu-l maxmood96 fanruliang borwe obiwahn arthur-s unpourtous ttldtor-dxfeed wujing-gxl dgrnbrg-meta yuriykulikov g0m3e kylesferrazza b-0n3 carun atry zhuang12346413 cvannost tspopp jell-o-fishi soso-p sigusr1

rsocket-cpp's Issues

Consider making ::onSubscribe optional

While maintaining the "unsubscribe handshake".

[feature request] simplify defining RequestHandler

Some possible options to make it simpler to use, especially around "kicking the tires" style functionality

A builder approach like the java where you standard with a concrete RequestHandler that rejects all requests and then add handlers
base class with concrete methods you can override
concrete utility classes for things like subscriptions with default behaviour e.g. when not implementing handleRequestChannel

Use the SubscriberBase/SubscriptionBase for implementing the automatons.

For this we need to work on the default executor. The current behavior would turn the regular callstack into unbounded queue of work and it would be extremely hard to debug. Performance would suffer too.

design an abstraction for single request handler

For scenarios when you want to implement ReactiveSocket proxy, it would be more efficient to work with API which would not allocate object for each stream, including all the overhead of smart pointers, etc.
The idea is to have an interface for having one request handler instance per reactive socket instance. All the current convenience classes like Subscriptions, Subscribers, etc. would be built atop this interface.

Create unit tests for TCP DuplexConnection

Warn when RequestHandler doesn't invoke response.onSubscribe()

Can we detect this after requestHandler.handle*() is called and print a warning or fail noisily and early? At least during development this should be the default.

production behaviour to be discussed.

propose new memory model

Over the last 6 months of using the api we gained some experience and these are some of the main drawbacks of such memory model:

memory management issues: it was very easy for the user to fall in a trap of "use-after-free" issues. We had several non-trivial bugs on iOS and fbcode.
the memory model was not obvious from the API signatures. It was hard for the newcomers to learn and understand the model especially around corner cases.
user had to use custom smart pointer wrappers (SubscriptionPtr, SubscriberPtr) to do the termination handshake right. Otherwise they would often cause memory corruption.
non-trivial chaining of Subscribers was hard to implement. This is mostly due to the fact that from any method the user can send a terminating signal (eq. from onSubscribe the user can call subscription::cancel) which would start destroying instances while their methods are still on the stack.
even simple unit tests had often memory bugs, especially use-after-free or memory corruption. This was mainly because the API wasn't clear as to whether the object has to be allocated on stack vs. heap.

New memory model is required.

Implement connection setup and connection-level automaton.

OpenSSL on OSX

I'm going to change the instructions for OSX development, but raising an issue for discussion.

I need to install openssl via brew and then run cmake like so

OPENSSL_ROOT_DIR=/usr/local/opt/openssl cmake ../ -DCMAKE_BUILD_TYPE=DEBUG

Is this the common experience?

Silence glog in test runs.

Test output is an unreadable crap, at the moment.

Lease support

currently not supported and not exposed in ConnectionSetupPayload

ConnectionAutomaton should ignore messages for streams it closed

I found this bug:
The server sends 2 payloads on a RS subscription. The client receives the first payload and cancels the subscription. Under the covers the client ConnectionAutomaton closes the RS stream. Then the second payload comes for the subscription. The ConnectoinAutomaton is now failing to find the stream and tries to handle the unknown stream and fails.

The core issues seems to be that there could be payloads to process between cancel and onComplete.

The fix should be either:

the ConnectionAutomaton ignores payloads for a closed stream, or
the underlying duplex connection will never try to deliver another payload after the the input subscription is canceled. I think this fix would not work in multithreaded environment, thou.

Configurable logging for LoggingMixin

I'm consuming ReactiveSockets in the ReactiveThrift framework. The LoggingMixin is a bit noisy, and so I've commented out the logging locally so that I can see my log statements. It would be great if this could be silent by default, and then the application could enable the ReactiveSockets logging, maybe through a FLAG_reactivesocket_logging_enabled = true;.

Make sure that there are no memory-management-related issues when no one calls Subscriber::onSubscribe.

This in conjunction with smart pointers being part of the interface gives quite strong exception safety during setup time. It also makes it much simpler to rely on smart pointers in general, as one can always put a Subscriber/Subscription in appropriate pointer even before ::onSubscribe has been called.

initial REQUEST_SUB/REQUEST_STREAM frame should not be sent until request(n) is called

today we are sending REQUEST_SUB frame and REQUEST frame separately. It really makes no sense to send REQUEST_SUB if the application code doesn't request responses

Configure clang-format.

post review of landed changes

@lehecka tagging this for a post review of #73 #74 #75 #76

But I've got a cycle in my workflow which I needed to break, e.g. we want a valid github SHA to update into internal repos.

Implement frame fragmentation

handle connection-level errors

handle connection-level errors (how to translate them into exceptions)

Socket termination should trigger onError for streams before RS.onClose

@benjchristensen

I had various subscriptions over a ReactiveSocket, then terminated the socket. Instead of the subscription streams receiving an 'onError' with an exception about the socket/connection closing, they received 'onComplete' and behaved as if everything had completed successfully.

It is my view that a stream should only receive an 'onComplete' if the origin on the server actually sent an onComplete and that ReactiveSocket itself can only terminate a stream via an 'onError'

TODO

confirm against java behaviour
update https://github.com/ReactiveSocket/reactivesocket/blob/master/Protocol.md
implement errors

Separate compilation for each automaton

They are all instantiated in ReactiveSocket.cpp right now, I believe.

Unnecessary check for COMPLETE flag in Frame_REQUEST_SUB/STREAM

Looking at SubscriptionResponderBase::onNextFrame(Frame_REQUEST_SUB&& frame) and StreamResponderBase::onNextFrame(Frame_REQUEST_STREAM&& frame), we’re checking if the Frame_REQUEST_SUB/STREAM frames received have the FrameFlags_COMPLETE flag set and if so we close the connection. However according to the protocol these two frame types don’t use the COMPLETE flag, and it is also never set by Stream/SubscriptionRequester.

Should these be removed?

Remove the use of SubscriberPtr and SubscriptionPtr

The smart pointers were protecting us from sending 2 terminating signals, mostly. We don't have that requirement now so we can remove them.

Please verify:

circular dependencies are taken care of
lifetime of instances while executing the methods is taken care of.

Once done, remove the smart pointers from the ReactiveStreams library.

Implement some intersection of transports with java impl

TCP at least

move AllowanceSemaphore.h from ReactiveStreams over to ReactiveSocket

Once done, lets remove the ReactiveStreams/utilities folder.

Payload missing metadata

Looks like the Payload object is missing a metadata field.

replace IOBuf with independent buffer implementation

this is along the lines removing folly dependency.

What are the options here?

cc: @tmontgomery

Library Size

The current compiled size of libReactiveSocket.a is 1.6M:

1.6M May 17 10:12 libReactiveSocket.a

The Java version is 185k and is feature complete:

185K May 17 10:14 reactivesocket-0.1.1-SNAPSHOT.jar

Java currently has a dependency on a 200k library, but that can likely be reduced with some effort.

So, reactivesocket-cpp is still not complete, and we are at 1.6M vs the complete Java impl being < 400k.

Folly is > 50M for everything, so I'm assuming the static linking is just pulling from it what is needed (showing my lack of C++ experience):

-rwxr-xr-x  1 root  wheel   5.9M Jan 11 02:15 /usr/local/lib/libfolly.57.dylib
-rw-r--r--  1 root  wheel    68M Jan 11 02:15 /usr/local/lib/libfolly.a

Is this greater size expected for C++? Or is it something we can reduce by being more careful with our dependencies?

I am exploring this due to desire to use this on mobile.

Document when to use reactivestreams::Subscriber vs SubscriberBase

We have both SubscriberBase and the reactivestreams Subscriber but it might not be clear how they are different, this should be documented.

ReactiveStreams interfaces should use smart pointers.

remove top level namespace

I'm happy to make the change, but want to agree the change first.

Are we ok with just dropping lithium and using reactivestreams and reactivesocket?

Set up contbuild.

I suppose we need to figure out how to build folly from source (or drop the dependency).

Allow destruction of ReactiveSocket instance from anywhere

...including any callbacks that are associated with the instance.

@yschimke @lehecka What should be the semantics over here?

Destroying ReactiveSocket kills all existing streams.
Destroying ReactiveSocket doesn't kill existing streams, but prevents any new streams from being accepted or requested.

Several mixins expect Subscriber<Payload> and don't appear to work with Subscriber<T> if T != Payload

For ReactiveThrift, I want to create a Subscriber<Foo>. A common pattern for subscribers appears to be to use createManagedInstance. This uses MemoryMixin, which defines

  void onNext(Payload payload) {

and so does not typecheck if my Subscriber defines

  void onNext(Foo payload) {

This might not be limited to just the MemoryMixin as a grep --include '*.h' Payload shows:

PublisherMixin.h:32:  void onNext(Payload payload) {
SinkIfMixin.h:21:class SinkIfMixin : public Base, public Subscriber<Payload> {
SinkIfMixin.h:29:  void onNext(Payload element) override final {
ExecutorMixin.h:50:  void subscribe(Subscriber<Payload>& subscriber) {
ExecutorMixin.h:78:  void onNext(Payload payload) {
ConsumerMixin.h:33:  void subscribe(Subscriber<Payload>& subscriber) {
ConsumerMixin.h:77:  reactivestreams::SubscriberPtr<Subscriber<Payload>> consumingSubscriber_;
MemoryMixin.h:46:  void subscribe(Subscriber<Payload>& subscriber) {
MemoryMixin.h:71:  void onNext(Payload payload) {
LoggingMixin.h:41:  void subscribe(Subscriber<Payload>& subscriber) {
LoggingMixin.h:67:  void onNext(Payload payload) {

Expose {Executor,Memory}Mixin with convenience wrappers.

Remove hard dependency on folly

Talking wth @benjchristensen, and second hand from @tmontgomery

We would ideally avoid having a hard dependency on folly. We should be able to build and run reactivesocket-cpp tests without folly. But then bind against folly for fb specific code.

Our TCP implementation might use folly AsyncSocket, but other implementations should be possible without building folly.

Q: Is there an open source alternative for our tests?
Q: What would Payload look like (since it is part of the ReactiveSocket public API), if an app wanted to use two transports one of which used folly:IOBuf e.g. proxygen and another used an open source alternative e.g. Aeron?

Make frame format compatible with protocol spec.

ExecutorMixin should cover signals sent by ConnectionAutomaton as well.

We have to order them w.r.t. signals sent by the clients, and the cleanest way to do it is to allow ExecutorMixin to handle it.

Crash in FrameBufferAllocator

[----------] 2 tests from ReactiveSocketTest
[ RUN      ] ReactiveSocketTest.RequestChannel
WARNING: Logging before InitGoogleLogging() is written to STDERR
I0506 16:37:32.444069 2101399552 LoggingMixin.h:44] ExecutorMixin(0x101a490f0, 2): subscribe(0x7fff5fbfde88)
I0506 16:37:32.444655 2101399552 LoggingMixin.h:44] MemoryMixin(0x101a490f0, 2): subscribe(0x7fff5fbfde88)
I0506 16:37:32.444663 2101399552 LoggingMixin.h:44] ChannelRequester(0x101a490f0, 2): subscribe(0x7fff5fbfde88)
I0506 16:37:32.444669 2101399552 LoggingMixin.h:44] ProducerMixin(0x101a490f0, 2): subscribe(0x7fff5fbfde88)
I0506 16:37:32.444672 2101399552 LoggingMixin.h:44] ConsumerMixin(0x101a490f0, 2): subscribe(0x7fff5fbfde88)
I0506 16:37:32.444705 2101399552 LoggingMixin.h:65] ExecutorMixin(0x101a490f0, 2): onSubscribe(0x7fff5fbfdc28)
I0506 16:37:32.444713 2101399552 LoggingMixin.h:65] MemoryMixin(0x101a490f0, 2): onSubscribe(0x7fff5fbfdc28)
I0506 16:37:32.444718 2101399552 LoggingMixin.h:65] ChannelRequester(0x101a490f0, 2): onSubscribe(0x7fff5fbfdc28)
I0506 16:37:32.444725 2101399552 LoggingMixin.h:65] ProducerMixin(0x101a490f0, 2): onSubscribe(0x7fff5fbfdc28)
I0506 16:37:32.444736 2101399552 LoggingMixin.h:70] ExecutorMixin(0x101a490f0, 2): onNext(<3>)
I0506 16:37:32.444748 2101399552 LoggingMixin.h:70] MemoryMixin(0x101a490f0, 2): onNext(<3>)
I0506 16:37:32.444754 2101399552 LoggingMixin.h:70] ChannelRequester(0x101a490f0, 2): onNext(<3>)
Process 53184 stopped
* thread #1: tid = 0x1211f92, 0x0000000101a0d080, queue = 'com.apple.main-thread', stop reason = EXC_BAD_ACCESS (code=2, address=0x101a0d080)
    frame #0: 0x0000000101a0d080
->  0x101a0d080: rcrb   $0x19, (%rdx)
    0x101a0d083: addb   %al, (%rcx)
    0x101a0d085: addb   %al, (%rax)
    0x101a0d087: addb   %al, (%rax)
(lldb) bt
* thread #1: tid = 0x1211f92, 0x0000000101a0d080, queue = 'com.apple.main-thread', stop reason = EXC_BAD_ACCESS (code=2, address=0x101a0d080)
  * frame #0: 0x0000000101a0d080
    frame #1: 0x0000000100098c98 ReactiveSocketTest`lithium::reactivesocket::FrameBufferAllocator::allocate(unsigned long) + 56
    frame #2: 0x0000000100099970 ReactiveSocketTest`lithium::reactivesocket::Frame_REQUEST_CHANNEL::serializeOut() + 64
    frame #3: 0x000000010008fb3d ReactiveSocketTest`void lithium::reactivesocket::ConnectionAutomaton::onNextFrame<lithium::reactivesocket::Frame_REQUEST_CHANNEL>(lithium::reactivesocket::Frame_REQUEST_CHANNEL&) + 61
    frame #4: 0x000000010008650f ReactiveSocketTest`lithium::reactivesocket::ChannelRequesterBase::onNext(std::__1::unique_ptr<folly::IOBuf, std::__1::default_delete<folly::IOBuf> >) + 1071
    frame #5: 0x00000001000b580e ReactiveSocketTest`lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::ChannelRequesterBase>::onNext(std::__1::unique_ptr<folly::IOBuf, std::__1::default_delete<folly::IOBuf> >) + 590
    frame #6: 0x00000001000b53bf ReactiveSocketTest`lithium::reactivesocket::MemoryMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::ChannelRequesterBase> >::onNext(std::__1::unique_ptr<folly::IOBuf, std::__1::default_d
elete<folly::IOBuf> >) + 303
    frame #7: 0x00000001000b504e ReactiveSocketTest`lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::MemoryMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::ChannelRequesterBase> > >::onNext(std::__1::uniq
ue_ptr<folly::IOBuf, std::__1::default_delete<folly::IOBuf> >) + 590
    frame #8: 0x00000001000b4c19 ReactiveSocketTest`lithium::reactivesocket::ExecutorMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::MemoryMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::ChannelRe
questerBase> > > >::onNext(std::__1::unique_ptr<folly::IOBuf, std::__1::default_delete<folly::IOBuf> >)::'lambda'()::operator()() + 345
    frame #9: 0x00000001000b4aad ReactiveSocketTest`void std::__1::__invoke_void_return_wrapper<void>::__call<lithium::reactivesocket::ExecutorMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::MemoryMixin<lithium::react
ivesocket::LoggingMixin<lithium::reactivesocket::ChannelRequesterBase> > > >::onNext(std::__1::unique_ptr<folly::IOBuf, std::__1::default_delete<folly::IOBuf> >)::'lambda'()&>(lithium::reactivesocket::ExecutorMixin<lithium::reactivesocket
::LoggingMixin<lithium::reactivesocket::MemoryMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::ChannelRequesterBase> > > >::onNext(std::__1::unique_ptr<folly::IOBuf, std::__1::default_delete<folly::IOBuf> >)::'lambda'(
)&&&) + 45
    frame #10: 0x00000001000b48f9 ReactiveSocketTest`std::__1::__function::__func<lithium::reactivesocket::ExecutorMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::MemoryMixin<lithium::reactivesocket::LoggingMixin<lith
ium::reactivesocket::ChannelRequesterBase> > > >::onNext(std::__1::unique_ptr<folly::IOBuf, std::__1::default_delete<folly::IOBuf> >)::'lambda'(), std::__1::allocator<lithium::reactivesocket::ExecutorMixin<lithium::reactivesocket::Logging
Mixin<lithium::reactivesocket::MemoryMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::ChannelRequesterBase> > > >::onNext(std::__1::unique_ptr<folly::IOBuf, std::__1::default_delete<folly::IOBuf> >)::'lambda'()>, void
()>::operator()() + 41
    frame #11: 0x0000000100332761 libfolly.48.dylib`folly::QueuedImmediateExecutor::addStatic(std::__1::function<void ()>) + 233
    frame #12: 0x00000001000b1f88 ReactiveSocketTest`void lithium::reactivesocket::ExecutorMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::MemoryMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::Cha
nnelRequesterBase> > > >::runInExecutor<lithium::reactivesocket::ExecutorMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::MemoryMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::ChannelRequesterBase>
 > > >::onNext(std::__1::unique_ptr<folly::IOBuf, std::__1::default_delete<folly::IOBuf> >)::'lambda'()>(lithium::reactivesocket::ExecutorMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::MemoryMixin<lithium::reactiveso
cket::LoggingMixin<lithium::reactivesocket::ChannelRequesterBase> > > >::onNext(std::__1::unique_ptr<folly::IOBuf, std::__1::default_delete<folly::IOBuf> >)::'lambda'()&&) + 696
    frame #13: 0x00000001000b1c42 ReactiveSocketTest`lithium::reactivesocket::ExecutorMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::MemoryMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::ChannelR
equesterBase> > > >::onNext(std::__1::unique_ptr<folly::IOBuf, std::__1::default_delete<folly::IOBuf> >) + 82
    frame #14: 0x00000001000b19ce ReactiveSocketTest`lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::ExecutorMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::MemoryMixin<lithium::reactivesocket::LoggingM
ixin<lithium::reactivesocket::ChannelRequesterBase> > > > >::onNext(std::__1::unique_ptr<folly::IOBuf, std::__1::default_delete<folly::IOBuf> >) + 590
    frame #15: 0x00000001000ada8f ReactiveSocketTest`lithium::reactivesocket::SinkIfMixin<lithium::reactivesocket::StreamIfMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::ExecutorMixin<lithium::reactivesocket::Logging
Mixin<lithium::reactivesocket::MemoryMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::ChannelRequesterBase> > > > > > >::onNext(std::__1::unique_ptr<folly::IOBuf, std::__1::default_delete<folly::IOBuf> >) + 303
    frame #16: 0x0000000100072ea4 ReactiveSocketTest`ReactiveSocketTest_RequestChannel_Test::TestBody()::$_1::operator()(unsigned long) const + 132
    frame #17: 0x0000000100072e17 ReactiveSocketTest`void testing::internal::InvokeHelper<void, std::__1::tuple<unsigned long> >::Invoke<ReactiveSocketTest_RequestChannel_Test::TestBody()::$_1>(ReactiveSocketTest_RequestChannel_Test::Test
Body()::$_1, std::__1::tuple<unsigned long> const&) + 55
    frame #18: 0x0000000100072dd4 ReactiveSocketTest`void testing::internal::InvokeAction<ReactiveSocketTest_RequestChannel_Test::TestBody()::$_1>::Perform<void, std::__1::tuple<unsigned long> >(std::__1::tuple<unsigned long> const&) + 52
    frame #19: 0x0000000100072d6c ReactiveSocketTest`testing::PolymorphicAction<testing::internal::InvokeAction<ReactiveSocketTest_RequestChannel_Test::TestBody()::$_1> >::MonomorphicImpl<void (unsigned long)>::Perform(std::__1::tuple<uns
igned long> const&) + 44
    frame #20: 0x0000000100011b21 ReactiveSocketTest`testing::Action<void (unsigned long)>::Perform(std::__1::tuple<unsigned long> const&) const + 257
    frame #21: 0x0000000100011db9 ReactiveSocketTest`testing::internal::ActionResultHolder<void>* testing::internal::ActionResultHolder<void>::PerformAction<void (unsigned long)>(testing::Action<void (unsigned long)> const&, testing::inte
rnal::Function<void (unsigned long)>::ArgumentTuple const&) + 25
    frame #22: 0x0000000100010b7c ReactiveSocketTest`testing::internal::FunctionMockerBase<void (unsigned long)>::UntypedPerformAction(void const*, void const*) const + 60
    frame #23: 0x0000000100152d83 ReactiveSocketTest`testing::internal::UntypedFunctionMockerBase::UntypedInvokeWith(void const*) + 7315
    frame #24: 0x0000000100016ba9 ReactiveSocketTest`testing::internal::FunctionMockerBase<void (unsigned long)>::InvokeWith(std::__1::tuple<unsigned long> const&) + 25
    frame #25: 0x0000000100016b85 ReactiveSocketTest`testing::internal::FunctionMocker<void (unsigned long)>::Invoke(unsigned long) + 197
    frame #26: 0x0000000100022565 ReactiveSocketTest`lithium::reactivestreams::UnmanagedMockSubscription::request_(unsigned long) + 69
    frame #27: 0x0000000100006f2d ReactiveSocketTest`lithium::reactivestreams::UnmanagedMockSubscription::request(unsigned long) + 29
    frame #28: 0x00000001000860bb ReactiveSocketTest`lithium::reactivesocket::ChannelRequesterBase::onSubscribe(lithium::reactivestreams::Subscription&) + 251
    frame #29: 0x00000001000b1754 ReactiveSocketTest`lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::ChannelRequesterBase>::onSubscribe(lithium::reactivestreams::Subscription&) + 196
    frame #30: 0x00000001000b1688 ReactiveSocketTest`lithium::reactivesocket::MemoryMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::ChannelRequesterBase> >::onSubscribe(lithium::reactivestreams::Subscription&) + 56
    frame #31: 0x00000001000b1624 ReactiveSocketTest`lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::MemoryMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::ChannelRequesterBase> > >::onSubscribe(lithium:
:reactivestreams::Subscription&) + 196
    frame #32: 0x00000001000b1558 ReactiveSocketTest`lithium::reactivesocket::ExecutorMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::MemoryMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::ChannelR
equesterBase> > > >::onSubscribe(lithium::reactivestreams::Subscription&) + 40
    frame #33: 0x00000001000b1504 ReactiveSocketTest`lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::ExecutorMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::MemoryMixin<lithium::reactivesocket::LoggingM
ixin<lithium::reactivesocket::ChannelRequesterBase> > > > >::onSubscribe(lithium::reactivestreams::Subscription&) + 196
    frame #34: 0x00000001000ad958 ReactiveSocketTest`lithium::reactivesocket::SinkIfMixin<lithium::reactivesocket::StreamIfMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::ExecutorMixin<lithium::reactivesocket::Logging
Mixin<lithium::reactivesocket::MemoryMixin<lithium::reactivesocket::LoggingMixin<lithium::reactivesocket::ChannelRequesterBase> > > > > > >::onSubscribe(lithium::reactivestreams::Subscription&) + 40
    frame #35: 0x000000010006d24e ReactiveSocketTest`ReactiveSocketTest_RequestChannel_Test::TestBody() + 8222
    frame #36: 0x0000000100144baa ReactiveSocketTest`testing::Test* testing::internal::HandleSehExceptionsInMethodIfSupported<testing::internal::TestFactoryBase, testing::Test*>(testing::internal::TestFactoryBase*, testing::Test* (testing
::internal::TestFactoryBase::*)(), char const*) + 122
    frame #37: 0x000000010012ebb7 ReactiveSocketTest`void testing::internal::HandleExceptionsInMethodIfSupported<testing::Test, void>(testing::Test*, void (testing::Test::*)(), char const*) + 119
    frame #38: 0x00000001000fd2a5 ReactiveSocketTest`testing::Test::Run() + 197
    frame #39: 0x00000001000fe978 ReactiveSocketTest`testing::TestInfo::Run() + 216
    frame #40: 0x00000001000ff607 ReactiveSocketTest`testing::TestCase::Run() + 231
    frame #41: 0x000000010010df0c ReactiveSocketTest`testing::internal::UnitTestImpl::RunAllTests() + 908
    frame #42: 0x0000000100145aea ReactiveSocketTest`bool testing::internal::HandleSehExceptionsInMethodIfSupported<testing::internal::UnitTestImpl, bool>(testing::internal::UnitTestImpl*, bool (testing::internal::UnitTestImpl::*)(), char
 const*) + 122
    frame #43: 0x0000000100130ba7 ReactiveSocketTest`bool testing::internal::HandleExceptionsInMethodIfSupported<testing::internal::UnitTestImpl, bool>(testing::internal::UnitTestImpl*, bool (testing::internal::UnitTestImpl::*)(), char co
nst*) + 119
    frame #44: 0x000000010010db28 ReactiveSocketTest`testing::UnitTest::Run() + 408
    frame #45: 0x0000000100165421 ReactiveSocketTest`RUN_ALL_TESTS() + 17
    frame #46: 0x0000000100165405 ReactiveSocketTest`main + 69
    frame #47: 0x00007fff923295ad libdyld.dylib`start + 1

Align reactive-streams-cpp classnames with Java

ReactiveStreams C++ version has different class names than the Java version. For instance a instead of a Publisher it has a Producer. Maybe I'm wrong but it seems like the C++ version should flow the naming convention in the Java implementation so it's easier to follow between them. The Processor class is missing as well.

DuplexConnection:onComplete should trigger onError on all streams.

When transport layer closes (cleanly or with error) all streams should be closed with onError because they didn't get a chance to close for themselfs.

Please note that this has to be eventually configurable as resumability comes along.

Simplify API for creating TCP client or server

Explain reasoning behind unsubscribe handshake.

folly (master) dependency on c++14

folly has been readded to brew. I was able to install locally from source.

brew install --build-from-source folly

However the reactivesocket-cpp build fails because of a lot of errors, first one specifically mentions a C++14 dep.

/usr/local/include/folly/Synchronized.h:157:3: error: 'auto' return without trailing return type; deduced return types are a C++14 extension
  auto withWLock(Function&& function) {
  ^

Editing CMakeLists.txt to use c++14 makes everything work again, but I believe we need c++11 support also.

remove copy constructor for MemoryMixin

I was doing the following, which compiled without optimisation.

auto input =
    *(new reactivesocket::MemoryMixin<PongSubscriberBase>(response));

Use interface / PIMPL for ReactiveSocket

Interface would be nice, as people can mock it.

final folder structure

We made some changes recently to the folder structure. We pulled reactive-streams-cpp to a separate repo, but kept the subfolder under reactivesocket-cpp. I believe it is just an obsolete artefact.
I want to understand the final folder structure. Here is my proposal:

reactive-streams-cpp (repo)

include
src
- utilities
test
examples
reactivesocket-cpp (repo)
src
- automata
- mixins
- streams
test
cmake
devtools
externals
...

The main difference between this proposal and the current state is essentially expanding reactive-streams-cpp/src to the root folder and moving reactivesocket-cpp/reactivesocket-cpp/src one level up.

Please let me know if there were any conversations about this which I missed.

cancel() doesn't work with `SubscriberBase`

The cancel() impl in SubscriptionShim first checks and swaps cancelled to true but then checks again, expecting it to be false.

See https://github.com/ReactiveSocket/reactivesocket-cpp/blob/master/src/SubscriberBase.h#L57

StreamCompletionSignal options should match spec

Filing so I don't forget.

We never use other values than GRACEFUL and ERROR from StreamCompletionSignal.
There is an assert here that can never trigger because of this: https://github.com/ReactiveSocket/reactivesocket-cpp/blob/master/src/ConnectionAutomaton.cpp#L105

For reference the current values of the enum are:

enum class StreamCompletionSignal {
  GRACEFUL,
  ERROR,
  INVALID_SETUP,
  UNSUPPORTED_SETUP,
  REJECTED_SETUP,
  CONNECTION_ERROR,
  CONNECTION_END,
};

Contended CAS on every buffer allocation

Originally, pre open-sourcing, the code used folly::Singleton::try_get, which is not available in the latest stable version of folly and incurs a contended CAS on every buffer allocation.
This have been replaced with folly::Singleton::get to fix build, but that change broke tests (why?) when compiling with CMake.
Once replaced back with get_weak().lock(), we again have a contended CAS on buffer allocation path. * For this reason, I'd vote to get rid of Singleton hack and inject a Singleton instance into frame serialisers.