René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Review and suggestions","articleBody":"Although I don't see any use in my work for such multi-observables currently, the idea about having higher arity dataflows made me interested (scientifically) in this library. Here, I'd like to write some points I see about such library.\n## Approach\n\nFirst, Instead of creating Bi- and Tri- Observables by hand, you can explore the possibility of code templating and static code generation, similar how many of the primitive-collections libraries (such as Trove or fastutil) do it.\n\nSecond, I suggest adopting the reactive-streams idioms instead of the RxJava 1.x idions. I don't see too much direct code reuse and since you'll need wrappers and converters anyway, you have the opportunity to avoid a few less-than-optimal decisions the RxJava API has made.\n## Primitives\n\nI'd start a new type hierarchy based on the reactive-streams idioms:\n\n``` java\ninterface {Arity}Publisher\u003cTi, ...\u003e {\n    void subscribe({Arity}Subscriber\u003cTi, ...\u003e s);\n}\ninterface SubscriberBase {\n    void onStart(Subscription s);\n    void onError(Throwable e);\n    void onCompleted();\n}\ninterface {Arity}Subscriber\u003cTi, ...\u003e extends SubscriberBase {\n    void onNext(T1 t1, ...);\n}\ninterface Subscription {  // or import reactive-streams directly\n    void request(long n);\n    void cancel();\n}\ninterface {Arity}Processor\u003cTi, ..., Ri, ...\u003e extends\n    {Arity}Subscriber\u003cTi,...\u003e, {Arity}Publisher\u003cRi, ...\u003e {\n} \n```\n## Queues\n\nSince most operators require some Spsc queue to operate and they are designed to transmit one element at a time, you'll need to extend the logic and API of the classical Queues.\n\n``` java\ninterface {Arity}Queue {\n    boolean offer(Ti, ...);\n    boolean poll(Action{k}\u003cTi, ...\u003e out);  // true if element was available\n    boolean peek(Action{k}\u003cTi, ...\u003e out);\n    boolean isEmpty();\n    int size();\n}\n```\n\nNote that indeed, using callbacks poses some overhead, but I think it is less than allocating some tuples for the classical queues.\n\nIn addition, the ring-buffer based Spsc queue implementation needs some slight changes regarding the element store: you can use the current indexing logic, but now that translates to k-times indexing into the underlying array (the power-of-2 remains), roughly:\n\n``` java\nObject[] array = new Object[powerOf2Capacity * 3];\n\nvoid offer(T1 t1, T2 t2, T3 t3) {\n    int offset = ((int)(producerIndex) \u0026 mask) * 3;\n    array[offset] = t1;\n    array[offset + 1] = t2;\n    array[offset + 2].lazySet(t3); // memory order: release\n    producerIndex++;\n}\nboolean poll(Action3\u003cT1, T2, T3\u003e out) {\n    int offset = ((int)(consumerIndex) \u0026 mask) * 3;\n    Object o3 = array[offset + 2]; // memory order: acquire\n    if (o1 == null { return false; }\n    Object o2 = array[offset + 1]; // read in opposite order\n    Object o1 = array[offset];\n    array[offset] = null;\n    array[offset + 1] = null;\n    array[offset + 2].lazySet(null);\n    consumerIndex++;\n    out.call((T1)o1, (T2)o2, (T3)o3);\n    return true;\n}\n```\n## Callbacks on the stream\n\nSince Java functions can't return more than one result, one can return a tuple for each 'column' and split the operators into many, or use the same trick as with queues and have an output callback:\n\n``` java\npublic final BiObservable\u003cR1, R2\u003e map(Action3\u003c? super T1, ? super T2, Action2\u003cR1, R2\u003e\u003e mapper) {\n    return lift(s -\u003e {\n        return new AbstractBiSubscriber\u003cT1, T2\u003e(s) {\n            @Override\n            public void onNext(T1 t1, T2 t2) {\n                mapper.call(t1, t2, s::onNext);\n            }\n            // ...\n        });\n    });\n}\n```\n","author":{"url":"https://github.com/akarnokd","@type":"Person","name":"akarnokd"},"datePublished":"2015-06-13T12:06:20.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":0},"url":"https://github.com/4/RxJavaMulti/issues/4"}