SDK: Upgrading to RxJava2

build.gradle

@@ -1,7 +1,7 @@
 // Top-level build file where you can add configuration options common to all sub-projects/modules.
 
 buildscript {
-    ext.kotlin_version = '1.1.3'
+    ext.kotlin_version = '1.1.3-2'
     ext.dokka_version = '0.9.14'
 
     repositories {

sdk/build.gradle

@@ -54,11 +54,12 @@ dependencies {
     // http request, api calls
     compile 'com.squareup.retrofit2:retrofit:2.3.0'
     compile 'com.squareup.retrofit2:converter-gson:2.3.0'
-    compile 'com.squareup.retrofit2:adapter-rxjava:2.3.0'
+    compile 'com.squareup.retrofit2:adapter-rxjava2:2.3.0'
     compile 'com.squareup.okhttp3:logging-interceptor:3.8.1'
     compile 'com.squareup.okhttp3:okhttp:3.8.1'
     compile 'com.google.code.gson:gson:2.8.0'
-    compile 'io.reactivex:rxandroid:1.2.1'
+    compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
+    compile 'io.reactivex.rxjava2:rxjava:2.1.1'
 
     androidTestCompile('com.android.support.test.espresso:espresso-core:2.2.2', {
         exclude group: 'com.android.support', module: 'support-annotations'

sdk/src/main/java/com/sygic/travel/sdk/StSDK.kt

@@ -17,11 +17,11 @@ import com.sygic.travel.sdk.model.place.Place
 import com.sygic.travel.sdk.model.place.Tour
 import com.sygic.travel.sdk.model.query.PlacesQuery
 import com.sygic.travel.sdk.model.query.ToursQuery
-import rx.Observable
-import rx.Scheduler
-import rx.Subscription
-import rx.android.schedulers.AndroidSchedulers
-import rx.schedulers.Schedulers
+import io.reactivex.Observable
+import io.reactivex.android.schedulers.AndroidSchedulers
+import io.reactivex.disposables.Disposable
+import io.reactivex.disposables.Disposables
+import io.reactivex.schedulers.Schedulers
 import java.io.File
 
 /**
@@ -62,7 +62,7 @@ class StSDK internal constructor() {
             }
         }
         val preparedObservable = getPreparedObservable(unpreparedObservable)
-        subscription = preparedObservable.subscribe(StObserver(callback, false))
+        disposable = preparedObservable.subscribeWith(StObserver(callback, false))
     }
 
     /**
@@ -85,7 +85,7 @@ class StSDK internal constructor() {
                 back.onFailure(t)
             }
         }
-        subscription = preparedObservable.subscribe(StObserver(callback, false))
+        disposable = preparedObservable.subscribeWith(StObserver(callback, false))
     }
 
     /**
@@ -108,7 +108,7 @@ class StSDK internal constructor() {
                 back.onFailure(t)
             }
         }
-        subscription = preparedObservable.subscribe(StObserver(callback, false))
+        disposable = preparedObservable.subscribeWith(StObserver(callback, false))
     }
 
     /**
@@ -135,16 +135,16 @@ class StSDK internal constructor() {
                 back?.onFailure(t)
             }
         }
-        subscription = preparedObservable.subscribe(StObserver(callback, false))
+        disposable = preparedObservable.subscribeWith(StObserver(callback, false))
     }
 
     /**
      *
      * Unsubscribes a subscribed observable.
      */
     fun unsubscribeObservable() {
-        if (subscription != null && !subscription!!.isUnsubscribed) {
-            subscription!!.unsubscribe()
+        if (!disposable.isDisposed) {
+            disposable.dispose()
         }
     }
 
@@ -154,7 +154,7 @@ class StSDK internal constructor() {
 
     private var stApi: StApi? = null
     private var cacheDir: File? = null
-    private var subscription: Subscription? = null
+    private var disposable : Disposable = Disposables.empty()
 
     /**
      *
@@ -184,11 +184,9 @@ class StSDK internal constructor() {
      * Prepares an [Observable] - sets [schedulers][Scheduler].
      * @param unpreparedObservable Observable to be prepared.
      * *
-     * @return Observable ready to be subscribed to.
+     * @return Disposable ready to be subscribed to.
      */
-    fun <T> getPreparedObservable(
-            unpreparedObservable: Observable<T>
-    ): Observable<T> {
+    fun <T> getPreparedObservable(unpreparedObservable: Observable<T>): Observable<T> {
         return unpreparedObservable
                 .subscribeOn(Schedulers.newThread())
                 .observeOn(AndroidSchedulers.mainThread())

sdk/src/main/java/com/sygic/travel/sdk/contentProvider/api/StApi.kt

@@ -1,11 +1,11 @@
 package com.sygic.travel.sdk.contentProvider.api
 
 import com.sygic.travel.sdk.api.responseWrappers.*
-import retrofit2.adapter.rxjava.Result
+import retrofit2.adapter.rxjava2.Result
 import retrofit2.http.GET
 import retrofit2.http.Path
 import retrofit2.http.Query
-import rx.Observable
+import io.reactivex.Observable
 
 /**
  *

sdk/src/main/java/com/sygic/travel/sdk/contentProvider/api/StApiGenerator.kt

@@ -10,7 +10,7 @@ import okhttp3.Cache
 import okhttp3.OkHttpClient
 import okhttp3.logging.HttpLoggingInterceptor
 import retrofit2.Retrofit
-import retrofit2.adapter.rxjava.RxJavaCallAdapterFactory
+import retrofit2.adapter.rxjava2.RxJava2CallAdapterFactory
 import retrofit2.converter.gson.GsonConverterFactory
 import java.io.File
 import java.util.concurrent.TimeUnit
@@ -54,7 +54,7 @@ object StApiGenerator {
     private val builder = Retrofit.Builder()
             .baseUrl(API_BASE_URL + VERSION_AND_LOCALE + "/")
             .addConverterFactory(GsonConverterFactory.create(apiGson))
-            .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
+            .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
 
     private fun getHttpClient(cacheDir: File): OkHttpClient? {
         if (httpClient == null) {

sdk/src/main/java/com/sygic/travel/sdk/contentProvider/api/StObserver.kt

@@ -2,8 +2,8 @@ package com.sygic.travel.sdk.contentProvider.api
 
 import com.sygic.travel.sdk.api.responseWrappers.StResponse
 import com.sygic.travel.sdk.api.responseWrappers.StResponse.Companion.STATUS_OK
-import retrofit2.adapter.rxjava.Result
-import rx.Observer
+import io.reactivex.observers.DisposableObserver
+import retrofit2.adapter.rxjava2.Result
 import java.io.IOException
 import java.util.*
 
@@ -22,7 +22,7 @@ class StObserver<RT : StResponse>
 (
         private val userCallback: Callback<RT>,
         private val multipleCallsMerged: Boolean
-) : Observer<Result<RT>> {
+) : DisposableObserver<Result<RT>>() {
 
     private var stResponse: RT? = null
     private val stResponses = ArrayList<RT?>()
@@ -32,7 +32,7 @@ class StObserver<RT : StResponse>
      * All API requests have been finished - not necessarily successfully. However an error could
      * have occurred, so the response must be checked.
      */
-    override fun onCompleted() {
+    override fun onComplete() {
         if (stResponse?.statusCode == STATUS_OK) {
             userCallback.onSuccess(stResponse!!)
         }