ReactiveX
diff --git a/‎src/main/java/io/reactivex/Flowable.java
Copy file name to clipboard
+1-1Lines changed: 1 addition & 1 deletion b/‎src/main/java/io/reactivex/Flowable.java
Copy file name to clipboard
+1-1Lines changed: 1 addition & 1 deletion
diff --git a/‎src/main/java/io/reactivex/Observable.java
Copy file name to clipboard
+1-1Lines changed: 1 addition & 1 deletion b/‎src/main/java/io/reactivex/Observable.java
Copy file name to clipboard
+1-1Lines changed: 1 addition & 1 deletion
diff --git a/‎src/main/java/io/reactivex/flowables/ConnectableFlowable.java
Copy file name to clipboard
+22-1Lines changed: 22 additions & 1 deletion b/‎src/main/java/io/reactivex/flowables/ConnectableFlowable.java
Copy file name to clipboard
+22-1Lines changed: 22 additions & 1 deletion
diff --git a/‎src/main/java/io/reactivex/internal/disposables/ResettableConnectable.java
Copy file name to clipboard
-54Lines changed: 0 additions & 54 deletions b/‎src/main/java/io/reactivex/internal/disposables/ResettableConnectable.java
Copy file name to clipboard
-54Lines changed: 0 additions & 54 deletions
@@ -12488,7 +12488,7 @@ public final <R> Flowable<R> publish(Function<? super Flowable<T>, ? extends Pub
     @SchedulerSupport(SchedulerSupport.NONE)
     public final ConnectableFlowable<T> publish(int bufferSize) {
         ObjectHelper.verifyPositive(bufferSize, "bufferSize");
-        return FlowablePublish.create(this, bufferSize);
+        return RxJavaPlugins.onAssembly(new FlowablePublish<T>(this, bufferSize));
     }
 
     /**
 
@@ -10207,7 +10207,7 @@ public final Observable<T> onTerminateDetach() {
     @CheckReturnValue
     @SchedulerSupport(SchedulerSupport.NONE)
     public final ConnectableObservable<T> publish() {
-        return ObservablePublish.create(this);
+        return RxJavaPlugins.onAssembly(new ObservablePublish<T>(this));
     }
 
     /**
 
@@ -34,11 +34,24 @@
  * before the {@code Flowable} begins emitting items.
  * <p>
  * <img width="640" height="510" src="https://github.com/ReactiveX/RxJava/wiki/images/rx-operators/publishConnect.png" alt="">
- *
+ * <p>
+ * When the upstream terminates, the {@code ConnectableFlowable} remains in this terminated state and,
+ * depending on the actual underlying implementation, relays cached events to late {@link Subscriber}s.
+ * In order to reuse and restart this {@code ConnectableFlowable}, the {@link #reset()} method has to be called.
+ * When called, this {@code ConnectableFlowable} will appear as fresh, unconnected source to new {@link Subscriber}s.
+ * Disposing the connection will reset the {@code ConnectableFlowable} to its fresh state and there is no need to call
+ * {@code reset()} in this case.
+ * <p>
+ * Note that although {@link #connect()} and {@link #reset()} are safe to call from multiple threads, it is recommended
+ * a dedicated thread or business logic manages the connection or resetting of a {@code ConnectableFlowable} so that
+ * there is no unwanted signal loss due to early {@code connect()} or {@code reset()} calls while {@code Subscriber}s are
+ * still being subscribed to to this {@code ConnectableFlowable} to receive signals from the get go.
+ * <p>
  * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Connectable-Observable-Operators">RxJava Wiki:
  *      Connectable Observable Operators</a>
  * @param <T>
  *          the type of items emitted by the {@code ConnectableFlowable}
+ * @since 2.0.0
  */
 public abstract class ConnectableFlowable<T> extends Flowable<T> {
 
@@ -53,6 +66,14 @@ public abstract class ConnectableFlowable<T> extends Flowable<T> {
      */
     public abstract void connect(@NonNull Consumer<? super Disposable> connection);
 
+    /**
+     * Resets this ConnectableFlowable into its fresh state if it has terminated.
+     * <p>
+     * Calling this method on a fresh or active {@code ConnectableFlowable} has no effect.
+     * @since 3.0.0
+     */
+    public abstract void reset();
+
     /**
      * Instructs the {@code ConnectableFlowable} to begin emitting the items from its underlying
      * {@link Flowable} to its {@link Subscriber}s.
Original file line number	Diff line number	Diff line change
`@@ -12488,7 +12488,7 @@ public final <R> Flowable<R> publish(Function<? super Flowable<T>, ? extends Pub`
`12488`	`12488`	`@SchedulerSupport(SchedulerSupport.NONE)`
`12489`	`12489`	`public final ConnectableFlowable<T> publish(int bufferSize) {`
`12490`	`12490`	`ObjectHelper.verifyPositive(bufferSize, "bufferSize");`
`12491`		`- return FlowablePublish.create(this, bufferSize);`
	`12491`	`+ return RxJavaPlugins.onAssembly(new FlowablePublish<T>(this, bufferSize));`
`12492`	`12492`	`}`
`12493`	`12493`
`12494`	`12494`	`/**`
Original file line number	Diff line number	Diff line change
`@@ -10207,7 +10207,7 @@ public final Observable<T> onTerminateDetach() {`
`10207`	`10207`	`@CheckReturnValue`
`10208`	`10208`	`@SchedulerSupport(SchedulerSupport.NONE)`
`10209`	`10209`	`public final ConnectableObservable<T> publish() {`
`10210`		`- return ObservablePublish.create(this);`
	`10210`	`+ return RxJavaPlugins.onAssembly(new ObservablePublish<T>(this));`
`10211`	`10211`	`}`
`10212`	`10212`
`10213`	`10213`	`/**`