make use BackPressureMode.BUFFER @ Observable creation @ MonitoringCSVSourceResourceTriplesFlow
*/
package org.dswarm.converter.flow;
-import java.io.Reader;
-import java.util.Collection;
-
import com.google.inject.Inject;
import com.google.inject.assistedinject.Assisted;
import org.culturegraph.mf.framework.ObjectPipe;
import org.culturegraph.mf.framework.ObjectReceiver;
import org.culturegraph.mf.types.Triple;
-import rx.Observable;
-import rx.Subscriber;
-
import org.dswarm.converter.DMPConverterException;
import org.dswarm.converter.mf.stream.converter.StreamToRecordTriples;
import org.dswarm.converter.mf.stream.reader.CsvReader;
@@ -33,6 +27,11 @@ import org.dswarm.converter.pipe.timing.ObjectTimer;
import org.dswarm.converter.pipe.timing.StreamTimer;
import org.dswarm.converter.pipe.timing.TimerBasedFactory;
import org.dswarm.persistence.model.resource.DataModel;
+import rx.Emitter;
+import rx.Observable;
+
+import java.io.Reader;
+import java.util.Collection;
/**
* @author phorn
@@ -66,15 +65,12 @@ public class MonitoringCSVSourceResourceTriplesFlow extends CSVSourceResourceTri
opener.setReceiver(csvReaderTimer).setReceiver(pipe);
- return Observable.create(new Observable.OnSubscribe<Collection<Triple>>() {
-
- @Override public void call(final Subscriber<? super Collection<Triple>> subscriber) {
+ return Observable.create(subscriber -> {
tripleReceiver.getObservable().subscribe(subscriber);
opener.process(obj);
opener.closeStream();
- }
- });
+ }, Emitter.BackpressureMode.BUFFER);
}
}
