官术网_书友最值得收藏!

Understanding the Observable.create method

You can create your own Observable with the Observable.create method at any time. This method takes an instance of the ObservableEmitter<T> interface as a source to observe on. So, let's consider this following example:

    fun main(args: Array<String>) { 
 
      val observer: Observer<String> = object : Observer<String> { 
        override fun onComplete() { 
          println("All Completed") 
        } 
 
        override fun onNext(item: String) { 
          println("Next $item") 
        } 
 
        override fun onError(e: Throwable) { 
          println("Error Occured ${e.message}") 
        } 
 
        override fun onSubscribe(d: Disposable) { 
          println("New Subscription ") 
        } 
    }//Create Observer 
 
    val observable:Observable<String> = Observable.create<String> {//1 
      it.onNext("Emit 1") 
      it.onNext("Emit 2") 
      it.onNext("Emit 3") 
      it.onNext("Emit 4") 
      it.onComplete() 
    } 
 
    observable.subscribe(observer) 
 
    val observable2:Observable<String> = Observable.create<String> {//2 
      it.onNext("Emit 1") 
      it.onNext("Emit 2") 
      it.onNext("Emit 3") 
      it.onNext("Emit 4") 
      it.onError(Exception("My Custom Exception")) 
    } 
 
    observable2.subscribe(observer) 
   } 

First, we created an instance of the Observer interface as the previous example. I will not elaborate on observer, as we have already seen an overview in the previous example, and we will see it in detail later in this chapter.

On comment 1, we created Observable with the Observable.create method; we emitted four string from Observable with the help of the onNext method, and then notified it is complete with the onComplete method.

On comment 2, we did almost the same, except here instead of calling onComplete, we called onError with a custom Exception.

Here is the output of the program:

The Observable.create method is useful, especially when you are working with a custom data structure and want to have control over what values are getting emitted. You can also emit values to Observer from a different thread.

Note that the Observable contract (http://reactivex.io/documentation/contract.html) states that Observable must issue notifications to observers serially (not in parallel). They may issue these notifications from different threads, but there must be a formal happens—before relationship between the notifications.

主站蜘蛛池模板: 宜兰市| 清苑县| 江油市| 舞钢市| 应城市| 信宜市| 白河县| 西乌珠穆沁旗| 贵南县| 稷山县| 长治市| 长武县| 永和县| 平陆县| 时尚| 云梦县| 集贤县| 临夏市| 中江县| 故城县| 揭东县| 香河县| 永川市| 涡阳县| 紫云| 普宁市| 墨江| 固阳县| 广安市| 伊吾县| 仁化县| 贵德县| 盐池县| 红桥区| 凌源市| 太仓市| 漳州市| 临高县| 武胜县| 乐陵市| 襄樊市|