Fix a race condition that may make OperatorMaterialize emit the

terminal notification more than once

The guards in `OperatorMaterialize.ParentSubscriber#drain` were never
working, because `busy` was actually never set to true.
Therefore it was possible that the `drain` loop was executed by
more than one thread concurrently, which could led to undefined
behavior.

This fix sets `busy` to true at the entry of `drain`.

Author: pkolaczk

src/main/java/rx/internal/operators/OperatorMaterialize.java

@@ -134,6 +134,7 @@ private void drain() {
                     missed = true;
                     return;
                 }
+                busy = true;
             }
             // drain loop
             final AtomicLong localRequested = this.requested;