3.7 Added serialize and unsafeSubscribe

Author: Froussios

Part 3 - Taming the sequence/7. Custom operators.md

@@ -269,6 +269,114 @@ If you can't guarantee that your operator will obey the Rx contract, for example
 
 ![](https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/synchronize.png)
 
+Let's first create an observable that breaks the contract and subscribe to it.
+
+```java
+Observable<Integer> source = Observable.create(o -> {
+	o.onNext(1);
+	o.onNext(2);
+	o.onCompleted();
+	o.onNext(3);
+	o.onCompleted();
+});
+
+source.subscribe(
+		System.out::println,
+		System.out::println,
+		() -> System.out.println("Completed"));
+```
+Output
+```
+1
+2
+Completed
+Unsubscribed
+```
+
+Despite what our observable is set to emit, the end result obeyed the Rx contract. That happened because `subscribe` will temrinate the subscription when the sequence terminates (or was supposed to). It doesn't mean that the problem will be taken care for us. There is also a method called `unsafeSubscribe`, which won't unsubscribe automatically.
+
+```java
+Observable<Integer> source = Observable.create(o -> {
+	o.onNext(1);
+	o.onNext(2);
+	o.onCompleted();
+	o.onNext(3);
+	o.onCompleted();
+});
+
+source.doOnUnsubscribe(() -> System.out.println("Unsubscribed"))
+	.unsafeSubscribe(new Subscriber<Integer>() {
+		@Override
+		public void onCompleted() {
+			System.out.println("Completed");
+		}
+
+		@Override
+		public void onError(Throwable e) {
+			System.out.println(e);
+		}
+
+		@Override
+		public void onNext(Integer t) {
+			System.out.println(t);
+		}
+});
+```
+Output
+```
+1
+2
+Completed
+3
+Completed
+```
+
+Our subscriber's behaviour was identical to the previous example (we created an instance of `Subscriber` because `unsafeSubscribe` doesn't have overloads that take lambdas). We can see here we weren't unsubscribed and we kept receiving notifications. 
+
+`unsafeSubscribe` is unsafe in other regards as well, such as error handling. It's usefulness is limited. The documentation says that it should only be used for custom operators that use nested subscriptions. To protect such operators from receiving and illegal sequence, we can apply the `serialize` operator
+
+```java
+Observable<Integer> source = Observable.create(o -> {
+		o.onNext(1);
+		o.onNext(2);
+		o.onCompleted();
+		o.onNext(3);
+		o.onCompleted();
+	})
+	.cast(Integer.class)
+	.serialize();;
+	
+
+source.doOnUnsubscribe(() -> System.out.println("Unsubscribed"))
+	.unsafeSubscribe(new Subscriber<Integer>() {
+		@Override
+		public void onCompleted() {
+			System.out.println("Completed");
+		}
+
+		@Override
+		public void onError(Throwable e) {
+			System.out.println(e);
+		}
+
+		@Override
+		public void onNext(Integer t) {
+			System.out.println(t);
+		}
+});
+```
+Output
+```
+1
+2
+Completed
+```
+
+We here that, despite the fact that we did not unsubscribe, the illegal notifications were filtered out.
+
+
+
+
 ### Extra benefits of lift
 
 If the ability to use your custom operator in the chain like a standard operator is not convincing enough, using `lift` has one more unexpected advantage. Standard operators are also implemented using `lift`, which makes `lift` a hot method at runtime. JVM optimises for `lift` and operators that use `lift` receive a performance boost. That can include your operator, if you use `lift`.