rx java - RxJava, OkHttp, Okio, file downloader in Android -

i trying convert working file download code reactive. got stuck, due little knowledge of rxjava. me make reactive?

public void downloadfile(messagecomponent media) {         request request = new request.builder()                 .url(media.getmediaurl())                 .build();          call call = http_client.newcall(request);         call.enqueue(new callback() {             @override             public void onfailure(request request, ioexception e) {                 log.e(tag, "failed execute " + request, e);             }              @override             public void onresponse(response response) throws ioexception {                 if (!response.issuccessful()) {                     throw new ioexception("unexpected code " + response);                 }                 string mimetype = mimetypemap.getfileextensionfromurl(media.getmediaurl());                 file file = new file(helper.gettmpfolder() + "/" + helper.generateuniquename() + "test." + mimetype);                 bufferedsink sink = okio.buffer(okio.sink(file));                 sink.writeall(response.body().source());                 sink.close();                 log.d(tag, "downloadfilefromserver done: " + media.getmediaurl());             }         });     }  

this have written far, , not getting result or error:

public void downloadfile(messagecomponent media){    observable<string> downloadobservable = observable.create(             sub -> {                 request request = new request.builder()                         .url(media.getmediaurl())                         .build();                 response response = null;                 try {                     response = http_client.newcall(request).execute();                     if (!response.issuccessful()) new ioexception();                 } catch (ioexception e) {                     e.printstacktrace();                 }                  sub.onnext(response.tostring());             }     );      subscriber<string> mysubscriber = new subscriber<string>() {         @override         public void onnext(string responsestring) {             log.d(tag, "works: " + responsestring);         }          @override         public void oncompleted() {         }          @override         public void onerror(throwable e) {             log.e(tag, e.getmessage(), e);         }     };     downloadobservable             .subscribeon(schedulers.newthread())             .observeon(androidschedulers.mainthread())             .subscribe(mysubscriber);     mysubscriber.unsubscribe();  } 

your code have errors, may explain expected behaviours not get.

observable contract error

reactive extensions (rxjava implemtation of it) based on contract : can notified multiple times on onnext then, notified once (or never...) on error or on completion.

onnext* (oncomplete | onerror)? 

so, observable code can rewriten this, emit fact stream in error or completed.

observable<string> downloadobservable = observable.create(         sub -> {             request request = new request.builder()                     .url(media.getmediaurl())                     .build();             response response = null;                 response = http_client.newcall(request).execute();                 if (response.issuccessful()) {                     sub.onnext(response.tostring());                     sub.oncompleted();                 } else {                     sub.onerror(new ioexception());                 }         } ); 

unsubscribe earlier

you unsubscribe after subscription, observable may not have time executed.

downloadobservable         .subscribeon(schedulers.newthread())         .observeon(androidschedulers.mainthread())         .subscribe(mysubscriber); mysubscriber.unsubscribe(); 

if observable complete, unsubscribe. won't have unsubscribe in case.