Always close requests' response body

app/src/main/java/eu/kanade/tachiyomi/data/mangasync/services/MyAnimeList.kt

@@ -77,8 +77,8 @@ class MyAnimeList(private val context: Context, id: Int) : MangaSyncService(cont
     }
 
     fun search(query: String): Observable<List<MangaSync>> {
-        return networkService.requestBody(get(getSearchUrl(query), headers))
-                .map { Jsoup.parse(it) }
+        return networkService.request(get(getSearchUrl(query), headers))
+                .map { Jsoup.parse(it.body().string()) }
                 .flatMap { Observable.from(it.select("entry")) }
                 .filter { it.select("type").text() != "Novel" }
                 .map {
@@ -102,8 +102,8 @@ class MyAnimeList(private val context: Context, id: Int) : MangaSyncService(cont
 
     // MAL doesn't support score with decimals
     fun getList(): Observable<List<MangaSync>> {
-        return networkService.requestBody(get(getListUrl(username), headers), networkService.forceCacheClient)
-                .map { Jsoup.parse(it) }
+        return networkService.request(get(getListUrl(username), headers), networkService.forceCacheClient)
+                .map { Jsoup.parse(it.body().string()) }
                 .flatMap { Observable.from(it.select("manga")) }
                 .map {
                     val manga = MangaSync.create(this)

app/src/main/java/eu/kanade/tachiyomi/data/network/NetworkHelper.kt

@@ -6,7 +6,9 @@ import okhttp3.OkHttpClient
 import okhttp3.Request
 import okhttp3.Response
 import rx.Observable
+import rx.subscriptions.Subscriptions
 import java.io.File
+import java.io.IOException
 
 class NetworkHelper(context: Context) {
 
@@ -39,14 +41,22 @@ class NetworkHelper(context: Context) {
         get() = cookieManager.store
 
     fun request(request: Request, client: OkHttpClient = defaultClient): Observable<Response> {
-        return Observable.fromCallable {
-            client.newCall(request).execute()
-        }
-    }
+        return Observable.create { subscriber ->
+            val call = client.newCall(request)
+            subscriber.add(Subscriptions.create { call.cancel() })
 
-    fun requestBody(request: Request, client: OkHttpClient = defaultClient): Observable<String> {
-        return Observable.fromCallable {
-            client.newCall(request).execute().body().string()
+            try {
+                val response = call.execute()
+                if (!subscriber.isUnsubscribed) {
+                    subscriber.add(Subscriptions.create { response.body().close() })
+                    subscriber.onNext(response)
+                    subscriber.onCompleted()
+                }
+            } catch (error: IOException) {
+                if (!subscriber.isUnsubscribed) {
+                    subscriber.onError(error)
+                }
+            }
         }
     }