reactivex / rxjavastring Goto Github PK

The StringObservable is in the same package as the regular Observable (rx.observables), so it isn't possible to import the package effectively.

Would it make sense to put it in its own package? I understand that changing the package is a big deal, as it breaks any consumer, but it is prettier in terms of modularisation.

Can you please extend the textual description in the README.md of what the library actually provides?

String operators for RxJava.

This appears a bit to short to me.

For people that uses Maven to try your code.

I'm doing:
StringObservable.byLine(StringObservable.decode(observableOfArrays, "UTF-8"))

The source observable handles backpressure and emits items only on demand, when they are requested. If lines of text span multiple arrays, processing of the stream stops after some time.

This is probably because the end subscriber requests 128 items on subscribe and this request gets passed upstream to the source observable without any modification. It seems no code in RxJava / RxJavaString takes into account the fact that the number of items produced by the source observable may be different than the number of lines finally produced at the end of the pipeline. In my case, because the source observable has to emit more arrays than 128 to build 128 lines, soon the end subscriber gets starved and processing stops.

At the moment if StringObservable.decode is called on a stream of byte[] and the source stream completes with bytes left over in the buffer (more is required before a string can be emitted) then the observable effectively hangs (onCompleted is not called). Because this situation cannot be distinguished from the source not completing or hanging I'd like to see StringObservable.decode call onError with some custom exception type that can be ignored if people wish using the normal means.

Are people ok with me submitting a PR for this?

It doesn't look like you're published yet

http://search.maven.org/#search%7Cga%7C1%7Cg%3A%22io.reactivex%22

Can we get a new release to Maven Central please.

We need to clean up the release tags as well. When we click on https://github.com/ReactiveX/RxJavaString/releases in github it indicates that the latest release is 0.21.0 when the latest release to Maven Central was 0.22.0. 1.0.0 is in the list of releases but is not in Maven Central.

Hi.
We are a Java 7 shop.

We are currently integrating StringObservable into our regular Observable builder flows by using compose and utility methods wrapping the Observables returned by StringObservable in Observable.Transformer instances. This is quite verbose. Is this the way StringObservables are currently intended to be integrated with regular Observable builder flows?

If the operators produced by StringObservable was also available (e.g. public static Observable.Operator<String, String> split(String regex) on a StringObservableOperators class) we could easily integrate StringObservable in our Observable builder flows with single liners using lift. E.g .lift(split(inputDelimiterRegex)).

Is that an addition you would consider?

Regards
Christian

@benjchristensen could you help turn off the Wiki for these RxJava*** projects? There are many spam wiki pages...

I looked at the following test:

@Test
public void testFromReaderWillUnsubscribeBeforeCallingNextRead() {
  final byte[] inBytes = "test".getBytes();
  final AtomicInteger numReads = new AtomicInteger(0);
  ByteArrayInputStream is = new ByteArrayInputStream(inBytes) {


    @Override
    public synchronized int read(byte[] b, int off, int len) {
      numReads.incrementAndGet();
      return super.read(b, off, len);
    }
  };
  StringObservable.from(new InputStreamReader(is)).first().toBlocking().single();
  assertEquals(1, numReads.get());
}

Looking at the test code, I don't see how this test is related to unsubscribe handling. All I see is a test that asserts that the overriden read() on the stream is called exactly once.

What am I missing?

I'm parsing a CSV file, and the byLine considers the whole file one long line.
The problem seems to be that the file lines use only CR line ends. (I don't know where that file came from)

The code looks like this:
public static Observable byLine(Observable source) {
return split(source, System.getProperty("line.separator"));
}

Using the system property for line separator isn't a very good idea, I think. It works when the file originates from the same platform, but it is a bit of a bugger to debug, and it can cause a 'works on my system' scenario.

I'd say just use the split and get rid of the byLine? Or add an explicit separator parameter (but then it is identical to split)

regards, Frank

https://github.com/ReactiveX/RxJavaString/blob/1.x/src/main/java/rx/observables/StringObservable.java#L425

If running on Windows (with line separator consisting of 2 chars \r\n), sometimes leftover can include full line with \r but no \n. During the next call, this left over is added AFTER the splits are processed. The effect is having 2 lines with line break in the first element.

To fix it, it needs to prepend leftover PRIOR to doing split.

As AsyncOnSubscribe has been removed from rx-java version 1.1.0 we are getting ClassNotFound exception when running with that version. The usage needs to be replaced with SyncOnSubscribe.

Hi there,

Issue #37 has been solved in March. I just pulled the latest version from Maven Central with the sole purpose of using .split, but suffered from this bug. Any plans on releasing master as a new version (and publishing to the central repo)?

Cheers,

Martijn

I'm ready to cut a 1.1.0 release of the library. If I don't at least two 👍 or an objection in a week I'll push it. If I do a few 👍 then I'll push it immediately.

• Pull 31 Make byLine handle carriage newline and just newline equally.
• Pull 36 update rx dependency to 1.1.1
• Pull 28 use the experimental version of using() that closes resources eagerly.

Artifacts: Maven Central

I used io.reactivex:rxandroid:0.24.0 and io.reactivex:rxjava-string:0.22.0

After ugrade dependency to io.reactivex:rxandroid:0.25.0, app became freezes on

public static String joinedBy(List<String> src, String separator) {
       return StringObservable.join(Observable.from(src), separator).toBlocking().single();
}

@davidmoten can we release a new version of RxJavaString?

Just run into a StackOverflow issue due to lack of backpressure support.

/cc @benjchristensen

I try to work with matchers over long string files... and experiment with this implementation.

example could be:

Pattern pattern = Pattern.compile("(<div id=\"(?<id>.*)\" data-template=\"(?<template>.*)\" data-src=\"(?<src>.*)\">(?<fallback>.*)<\\/div>)");

String content = "blabla....";

Observable<Map> matcherMap = StringObservable.matcher(content, pattern,"id","template","src","fallback");

Implantation:

public static Observable<Map> matcher(final Observable<String> src, final Pattern pattern, final String... names) {
        return src.lift(new Operator<Map, String>() {
            @Override
            public Subscriber<? super String> call(final Subscriber<? super Map> subscriber) {

                return new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        if (!subscriber.isUnsubscribed())
                            subscriber.onCompleted();
                    }

                    @Override
                    public void onError(final Throwable e) {
                        if (!subscriber.isUnsubscribed())
                            subscriber.onError(e);
                    }

                    @Override
                    public void onNext(final String segment) {
                        final Matcher matcher = pattern.matcher(segment);
                        while (!subscriber.isUnsubscribed() && matcher.find()) {
                            Map matcherMap = new HashMap();
                            for (String name : Arrays.asList(names)) {
                               matcherMap.put(name,matcher.group(name));
                            }
                            subscriber.onNext(matcherMap);
                        }
                    }
                };
            }
        });
    }

Running

./gradlew

gives

Error: Could not find or load main class org.gradle.wrapper.GradleWrapperMain

Can we get a new release of rxjava-string? I'd like to have the NPE fix in PR #4 in an artifact.