# Writing a custom reactive base type

DevFeed: [Writing a custom reactive base type](<https://devfeed.tech/articles/writing-a-custom-reactive-base-type-24800.md>)

Original publisher: [Read original article](<https://akarnokd.blogspot.com/2016/03/writing-custom-reactive-base-type.html>)

Author: David Karnok (noreply@blogger.com)

Published: 2016-03-20T13:18:00Z

Content type: tutorial

Language: en

Sources: [Akarnokd - Advanced RxJava](<https://devfeed.tech/sources/akarnokd-advanced-rxjava.md>)

Topics: [reactive](<https://devfeed.tech/topics/reactive.md>), [Java](<https://devfeed.tech/topics/java.md>), [Programming](<https://devfeed.tech/topics/programming.md>), [Code](<https://devfeed.tech/topics/code.md>)

Tags: [code](<https://devfeed.tech/tags/code.md>), [java](<https://devfeed.tech/tags/java.md>), [programming](<https://devfeed.tech/tags/programming.md>), [reactive](<https://devfeed.tech/tags/reactive.md>), [rxjava](<https://devfeed.tech/tags/rxjava.md>)

## AI overview

This tutorial explains how to create a custom reactive base type around RxJava's Observable. It shows how wrapping can add custom operators, hide unsuitable operators, and provide a distinct type for specialized processing pipelines, while also discussing interoperability with other observable types.

## Source excerpt

Introduction From time to time, the question or request comes up that one would really like to have his/her own reactive type. Even though RxJava's Observable has plenty of methods and extension points via lift(), extend() and compose(), one feels the Observable should have the operator xyz() or in some chains, the chain shouldn't allow calling uvw(). The first case, namely adding a new custom method without going through the project as a contribution, is as old as the reactive programming on the JVM. When I first ported Rx.NET to Java, I had to face the same problem because .NET had the very convenient extension method support already back in 2010. Java doesn't have this and the idea has been rejected in the version 8 development era in the favor of default methods with the "justification" that such extension methods can't be overridden. Yes they can't but they can be replaced by another method from another class. The second case, hiding or removing operators, comes up with custom Observables where certain operations don't make sense. For example, given a ParallelObservable that splits the input sequence into parallel processing pipelines internally, it makes sense to map() or filter() in parallel, but it doesn't make sense to use take() or skip(). Wrapping Both cases can be solved by writing a custom type and just wrap the Observable into it. public final class MyObservable<T> { private Observable<T> actual; public MyObservable<T>(Observable<T> actual) { this.actual = actual; } } Now we can add operators of our liking: // ... public static <T> MyObservable<T> create(Observable<T> o) { return new MyObservable<T>(o); } public static <T> MyObservable<T> just(T value) { return create(Observable.just(value)); } public final MyObservable<T> goAsync() { return create(actual.subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread())); } public final <R> MyObservable<R> map(Func1<T, R> mapper) { return create(actual.map(mapper)); } public final void subscribe(