Fix FlowableOnBackpressureBufferStrategy

[matgabriel]

Fix buffered objects not propagated downstream in FlowableOnBackpressureBufferStrategy
Fix drop strategy logic in FlowableOnBackpressureBufferStrategy
Add unit test for FlowableOnBackpressureBufferStrategy, copied from FlowableOnBackpressureBufferTest, there is still some work needed to have a better coverage

[akarnokd]

Use Functions.EMPTY_ACTION.

[codecov-io]

Current coverage is 75.32% (diff: 66.66%)

Merging #4441 into 2.x will increase coverage by 0.17%

@@                2.x      #4441   diff @@
==========================================
  Files           454        454          
  Lines         32271      32274     +3   
  Methods           0          0          
  Messages          0          0          
  Branches       5208       5209     +1   
==========================================
+ Hits          24249      24309    +60   
+ Misses         6002       5928    -74   
- Partials       2020       2037    +17   

Powered by Codecov. Last update a856572...1617087

[akarnokd]

Should not call drain while in the synchronized block. It should look like this:

            if (done) {
                return;
            }
            boolean callOnOverflow = false;
            boolean callError = false;
            Deque<T> dq = deque;
            synchronized (dq) {
               if (dq.size() == bufferSize) {
                   switch (strategy) {
                   case DROP_LATEST:
                       dq.pollLast();
                       dq.offer(t);
                       callOnOverflow = true;
                       break;
                   case DROP_OLDEST:
                       dq.poll();
                       dq.offer(t);
                       callOnOverflow = true;
                       break;
                   default:
                       // signal error
                       callError = true;
                       break;
                   }
               } else {
                   dq.offer(t);
                   return;
               }
            }

            if (callOnOverflow) {
                if (onOverflow != null) {
                    try {
                        onOverflow.run();
                    } catch (Throwable ex) {
                        Exceptions.throwIfFatal(ex);
                        s.cancel();
                        onError(ex);
                    }
                }
            } 
            else if (callError) {
                s.cancel();
                onError(new MissingBackpressureException());
            } else {
                drain();
            }

[akarnokd]

Thanks! 👍