Implement 'toString' method for some Emitters

src/main/java/io/reactivex/internal/operators/completable/CompletableCreate.java

@@ -118,5 +118,10 @@ public void dispose() {
         public boolean isDisposed() {
             return DisposableHelper.isDisposed(get());
         }
+
+        @Override
+        public String toString() {
+            return String.format("%s{%s}", getClass().getSimpleName(), super.toString());
+        }
     }
 }

src/main/java/io/reactivex/internal/operators/flowable/FlowableCreate.java

@@ -233,6 +233,11 @@ public boolean isCancelled() {
         public FlowableEmitter<T> serialize() {
             return this;
         }
+
+        @Override
+        public String toString() {
+            return emitter.toString();
+        }
     }
 
     abstract static class BaseEmitter<T>
@@ -338,6 +343,11 @@ public final long requested() {
         public final FlowableEmitter<T> serialize() {
             return new SerializedEmitter<T>(this);
         }
+
+        @Override
+        public String toString() {
+            return String.format("%s{%s}", getClass().getSimpleName(), super.toString());
+        }
     }
 
     static final class MissingEmitter<T> extends BaseEmitter<T> {

src/main/java/io/reactivex/internal/operators/maybe/MaybeCreate.java

@@ -145,5 +145,10 @@ public void dispose() {
         public boolean isDisposed() {
             return DisposableHelper.isDisposed(get());
         }
+
+        @Override
+        public String toString() {
+            return String.format("%s{%s}", getClass().getSimpleName(), super.toString());
+        }
     }
 }

src/main/java/io/reactivex/internal/operators/observable/ObservableCreate.java

@@ -126,6 +126,11 @@ public void dispose() {
         public boolean isDisposed() {
             return DisposableHelper.isDisposed(get());
         }
+
+        @Override
+        public String toString() {
+            return String.format("%s{%s}", getClass().getSimpleName(), super.toString());
+        }
     }
 
     /**
@@ -279,6 +284,11 @@ public boolean isDisposed() {
         public ObservableEmitter<T> serialize() {
             return this;
         }
+
+        @Override
+        public String toString() {
+            return emitter.toString();
+        }
     }
 
 }

src/main/java/io/reactivex/internal/operators/single/SingleCreate.java

@@ -123,5 +123,10 @@ public void dispose() {
         public boolean isDisposed() {
             return DisposableHelper.isDisposed(get());
         }
+
+        @Override
+        public String toString() {
+            return String.format("%s{%s}", getClass().getSimpleName(), super.toString());
+        }
     }
 }

src/test/java/io/reactivex/internal/operators/completable/CompletableCreateTest.java

@@ -297,4 +297,14 @@ public void subscribe(CompletableEmitter e) throws Exception {
             RxJavaPlugins.reset();
         }
     }
+
+    @Test
+    public void emitterHasToString() {
+        Completable.create(new CompletableOnSubscribe() {
+            @Override
+            public void subscribe(CompletableEmitter emitter) throws Exception {
+                assertTrue(emitter.toString().contains(CompletableCreate.Emitter.class.getSimpleName()));
+            }
+        }).test().assertEmpty();
+    }
 }

src/test/java/io/reactivex/internal/operators/flowable/FlowableCreateTest.java

@@ -16,7 +16,9 @@
 import static org.junit.Assert.*;
 
 import java.io.IOException;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 import org.junit.Test;
 import org.reactivestreams.*;
@@ -931,4 +933,26 @@ public void subscribe(FlowableEmitter<Object> e) throws Exception {
             }
         }
     }
+
+    @Test
+    public void emittersHasToString() {
+        Map<BackpressureStrategy, Class<? extends FlowableEmitter>> emitterMap =
+                new HashMap<BackpressureStrategy, Class<? extends FlowableEmitter>>();
+
+        emitterMap.put(BackpressureStrategy.MISSING, FlowableCreate.MissingEmitter.class);
+        emitterMap.put(BackpressureStrategy.ERROR, FlowableCreate.ErrorAsyncEmitter.class);
+        emitterMap.put(BackpressureStrategy.DROP, FlowableCreate.DropAsyncEmitter.class);
+        emitterMap.put(BackpressureStrategy.LATEST, FlowableCreate.LatestAsyncEmitter.class);
+        emitterMap.put(BackpressureStrategy.BUFFER, FlowableCreate.BufferAsyncEmitter.class);
+
+        for (final Map.Entry<BackpressureStrategy, Class<? extends FlowableEmitter>> entry : emitterMap.entrySet()) {
+            Flowable.create(new FlowableOnSubscribe<Object>() {
+                @Override
+                public void subscribe(FlowableEmitter<Object> emitter) throws Exception {
+                    assertTrue(emitter.toString().contains(entry.getValue().getSimpleName()));
+                    assertTrue(emitter.serialize().toString().contains(entry.getValue().getSimpleName()));
+                }
+            }, entry.getKey()).test().assertEmpty();
+        }
+    }
 }

src/test/java/io/reactivex/internal/operators/maybe/MaybeCreateTest.java

@@ -335,4 +335,14 @@ public void subscribe(MaybeEmitter<Object> e) throws Exception {
             RxJavaPlugins.reset();
         }
     }
+
+    @Test
+    public void emitterHasToString() {
+        Maybe.create(new MaybeOnSubscribe<Object>() {
+            @Override
+            public void subscribe(MaybeEmitter<Object> emitter) throws Exception {
+                assertTrue(emitter.toString().contains(MaybeCreate.Emitter.class.getSimpleName()));
+            }
+        }).test().assertEmpty();
+    }
 }

src/test/java/io/reactivex/internal/operators/observable/ObservableCreateTest.java

@@ -643,4 +643,15 @@ public void subscribe(ObservableEmitter<Object> e) throws Exception {
             RxJavaPlugins.reset();
         }
     }
+
+    @Test
+    public void emitterHasToString() {
+        Observable.create(new ObservableOnSubscribe<Object>() {
+            @Override
+            public void subscribe(ObservableEmitter<Object> emitter) throws Exception {
+                assertTrue(emitter.toString().contains(ObservableCreate.CreateEmitter.class.getSimpleName()));
+                assertTrue(emitter.serialize().toString().contains(ObservableCreate.CreateEmitter.class.getSimpleName()));
+            }
+        }).test().assertEmpty();
+    }
 }

src/test/java/io/reactivex/internal/operators/single/SingleCreateTest.java

@@ -307,4 +307,14 @@ public void subscribe(SingleEmitter<Object> e) throws Exception {
             RxJavaPlugins.reset();
         }
     }
+
+    @Test
+    public void emitterHasToString() {
+        Single.create(new SingleOnSubscribe<Object>() {
+            @Override
+            public void subscribe(SingleEmitter<Object> emitter) throws Exception {
+                assertTrue(emitter.toString().contains(SingleCreate.Emitter.class.getSimpleName()));
+            }
+        }).test().assertEmpty();
+    }
 }