java - Convert String to Array to Objects in Observable -

i'm trying use closeablehttpasyncclient read endpoint, marshall string object (using javax.json) convert array on object it's individual components:

closeablehttpasyncclient client = httpasyncclientbuilder.create().setdefaultcredentialsprovider(provider).build();  client.start();  observable<observablehttpresponse> observable = observablehttp.createrequest(httpasyncmethods.createget(uri), client)             .toobservable();  observable<jsonarray> shareable = observable.flatmap(response -> response.getcontent().map(bb -> {         string stringval = new string(bb);         stringreader reader = new stringreader(stringval);         jsonobject jobj = json.createreader(reader).readobject();         return jobj.getjsonarray("elements");     })).share(); 

i need json array, filter objects of array:

observable<jsonobject> firststream = shareable.filter(item -> item.getstring("type").equals("type_1")); observable<jsonobject> secondstream = shareable.filter(item -> item.getstring("type").equals("type_2")); observable<jsonobject> thirdstream = shareable.filter(item -> item.getstring("type").equals("type_3")); 

how convert observable<jsonarray> observablejsonobject>?

because it's async, can't use foreach create kind of array buffer data.

update:

so looking @ using closeablehttpasyncclient may not best solution i'm trying achieve - realised morning (in shower of things) i'm trying process data asynchronously make async calls.

ideally, making call closeablehttpclient (sync) , passing data observable filtering more ideal approach (i don't need first call manage more 1 http call).

    closeablehttpclient client = httpclientbuilder.create().setdefaultcredentialsprovider(provider).build();      stringbuffer result = new stringbuffer();      try {         httpget request = new httpget(uri);         httpresponse response = client.execute(request);          bufferedreader rd = new bufferedreader(                 new inputstreamreader(response.getentity().getcontent()));          string line;         while ((line = rd.readline()) != null) {             result.append(line);         }     } catch(clientprotocolexception cpe) { } catch(ioexception ioe) {  }      stringreader reader = new stringreader(result.tostring());     jsonobject jobj = json.createreader(reader).readobject();     jsonarray elements = jobj.getjsonarray("elements");      list<jsonobject> objects = elements.getvaluesas(jsonobject.class);       observable<jsonobject> shareable = observable.from(objects).share();      observable<jsonobject> firststream = shareable.filter(item -> item.getstring("type").equals("type_1"));     observable<jsonobject> secondstream = shareable.filter(item -> item.getstring("type").equals("type_2"));     observable<jsonobject> thirdstream = shareable.filter(item -> item.getstring("type").equals("type_3"));       firststream.subscribe(record -> {         //connect sots/facebook , store results         bytearrayoutputstream baos = new bytearrayoutputstream();         json.createwriter(baos).writeobject(record);         system.out.println(baos.tostring());     });      secondstream.subscribe(record -> {         bytearrayoutputstream baos = new bytearrayoutputstream();         json.createwriter(baos).writeobject(record);         system.out.println(baos.tostring());     });      thirdstream.subscribe(record -> {         bytearrayoutputstream baos = new bytearrayoutputstream();         json.createwriter(baos).writeobject(record);         system.out.println(baos.tostring());     }); 

try code:

    string myjson = "{\"elements\": [{\"text\":\"obj1\"},{\"text\":\"obj2\"}, {\"text\":\"obj3\"}]}";      observable.just(myjson)             .map(jsonstr -> new stringreader(myjson))             .map(reader -> json.createreader(reader).readobject())             .map(jobj -> jobj.getjsonarray("elements"))             .map(elements -> elements.toarray(new jsonobject[elements.size()]))             .flatmap(jsonobjects -> observable.from(jsonobjects))             .subscribe(                     (jsonobject) -> system.out.println(jsonobject.getstring("text")),                     throwable -> throwable.printstacktrace(),                     () -> system.out.println("on complete")); 

result:

07-22 12:19:22.362 8032-8032/com.mediamanagment.app i/system.out﹕ obj1
07-22 12:19:22.362 8032-8032/com.mediamanagment.app i/system.out﹕ obj2
07-22 12:19:22.362 8032-8032/com.mediamanagment.app i/system.out﹕ obj3

note:
should use dependency:

compile 'org.glassfish:javax.json:1.0.4' 

instead this:

compile 'javax.json:javax.json-api:1.0' 

if use 'javax.json:javax.json-api:1.0' javax.json.jsonexception: provider org.glassfish.json.jsonproviderimpl not found @ step:

.map(reader -> json.createreader(reader).readobject()) 

thereof, please, use 'org.glassfish:javax.json:1.0.4'

update: also, instead of

.flatmap(jsonobjects -> observable.from(jsonobjects)) 

you can use flatmapiterable( ):

.flatmapiterable(jsonobjects -> jsonobjects)