{"action":"created","issue":{"url":"https://api.github.com/repos/ReactiveX/RxJava/issues/2641","labels_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/2641/labels{/name}","comments_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/2641/comments","events_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/2641/events","html_url":"https://github.com/ReactiveX/RxJava/pull/2641","id":56974836,"number":2641,"title":"Experimental Proposal of rx.Task","user":{"login":"benjchristensen","id":813492,"avatar_url":"https://avatars.githubusercontent.com/u/813492?v=3","gravatar_id":"","url":"https://api.github.com/users/benjchristensen","html_url":"https://github.com/benjchristensen","followers_url":"https://api.github.com/users/benjchristensen/followers","following_url":"https://api.github.com/users/benjchristensen/following{/other_user}","gists_url":"https://api.github.com/users/benjchristensen/gists{/gist_id}","starred_url":"https://api.github.com/users/benjchristensen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/benjchristensen/subscriptions","organizations_url":"https://api.github.com/users/benjchristensen/orgs","repos_url":"https://api.github.com/users/benjchristensen/repos","events_url":"https://api.github.com/users/benjchristensen/events{/privacy}","received_events_url":"https://api.github.com/users/benjchristensen/received_events","type":"User","site_admin":false},"labels":[{"url":"https://api.github.com/repos/ReactiveX/RxJava/labels/Enhancement","name":"Enhancement","color":"84b6eb"}],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":35,"created_at":"2015-02-08T23:54:34Z","updated_at":"2015-03-19T12:46:04Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/2641","html_url":"https://github.com/ReactiveX/RxJava/pull/2641","diff_url":"https://github.com/ReactiveX/RxJava/pull/2641.diff","patch_url":"https://github.com/ReactiveX/RxJava/pull/2641.patch"},"body":"Adds `rx.Task` as a \"scalar Observable\" for representing work with a single return value.\r\n\r\nSee https://github.com/ReactiveX/RxJava/issues/1594 rx.Future/Task\r\n\r\nThis provides a type similar to `Future` in that it represents a scalar unit of work, but it is lazy like an `Observable` and many `Task`s can be combined into an `Observable` stream. Note how `Task.zip` returns `Task<R>` whereas `Task.merge` returns `Observable<R>`.\r\n\r\nNOTE: This is for experimentation and feedback at this time.\r\n\r\nItems requiring review and work that I'm particularly aware of:\r\n\r\n- naming of `OnExecute`\r\n- naming of `TaskObserver` (this one in particular I don't like)\r\n- design and implementation of `Task.Promise`\r\n- should the public `lift` use the `Observable.Operator` or should that only be for internal reuse?\r\n- should we have a public `lift` that uses a `Task.Operator`?\r\n- the `Task.toObservable` implementation right now is efficient but will likely break something so it likely needs to change to use `subscribe`\r\n- implementation of this merge variant: `Task<T> merge(final Task<? extends Task<? extends T>> source)`\r\n- several operators currently just wrap as `Observable` to reuse existing operators ... is that okay performance wise?\r\n- Javadocs\r\n\r\nExamples of using this class:\r\n\r\n```java\r\nimport rx.Observable;\r\nimport rx.Task;\r\nimport rx.Task.Promise;\r\n\r\npublic class TaskExamples {\r\n\r\n    public static void main(String... args) {\r\n        // scalar synchronous value\r\n        Task<String> t1 = Task.create(t -> {\r\n            t.onSuccess(\"Hello World!\");\r\n        });\r\n\r\n        // scalar synchronous value using helper method\r\n        Task<Integer> t2 = Task.just(1);\r\n\r\n        // synchronous error\r\n        Task<String> error = Task.create(t -> {\r\n            t.onError(new RuntimeException(\"failed!\"));\r\n        });\r\n\r\n        // executing\r\n        t1.subscribe(System.out::println);\r\n        t2.subscribe(System.out::println);\r\n        error.subscribe(System.out::println, e -> System.out.println(e.getMessage()));\r\n\r\n        // scalar Tasks for request/response like a Future\r\n        getData(1).subscribe(System.out::println);\r\n        getDataUsingPromise(2).subscribe(System.out::println);\r\n\r\n        // combining Tasks into another Task\r\n        Task<String> zipped = Task.zip(t1, t2, (a, b) -> a + \" -- \" + b);\r\n\r\n        // combining Tasks into an Observable stream\r\n        Observable<String> merged = Task.merge(t1, t2.map(String::valueOf), getData(3));\r\n        Observable<String> mergeWith = t1.mergeWith(t2.map(String::valueOf));\r\n\r\n        zipped.subscribe(v -> System.out.println(\"zipped => \" + v));\r\n        merged.subscribe(v -> System.out.println(\"merged => \" + v));\r\n        mergeWith.subscribe(v -> System.out.println(\"mergeWith => \" + v));\r\n    }\r\n\r\n    /**\r\n     * Example of an async scalar execution using Task.create\r\n     * <p>\r\n     * This shows the lazy, idiomatic approach for Rx exactly like an Observable except scalar.\r\n     *\r\n     * @param arg\r\n     * @return\r\n     */\r\n    public static Task<String> getData(int arg) {\r\n        return Task.create(s -> {\r\n            new Thread(() -> {\r\n                try {\r\n                    Thread.sleep(500);\r\n                } catch (Exception e) {\r\n                    e.printStackTrace();\r\n                }\r\n                // deliver value\r\n                    s.onSuccess(\"Data=\" + arg);\r\n                }).start();\r\n        });\r\n    }\r\n\r\n    /**\r\n     * Example of an async scalar execution using a Task.Promise\r\n     * <p>\r\n     * This shows how an eager (hot) process would work like using a Future.\r\n     *\r\n     * @param arg\r\n     * @return\r\n     */\r\n    public static Task<String> getDataUsingPromise(int arg) {\r\n        Task.Promise<String> p = Promise.create();\r\n\r\n        new Thread(() -> {\r\n            try {\r\n                Thread.sleep(500);\r\n            } catch (Exception e) {\r\n                e.printStackTrace();\r\n            }\r\n            // deliver value\r\n                p.onSuccess(\"Data=\" + arg);\r\n            }).start();\r\n\r\n        return p.getTask();\r\n    }\r\n}\r\n```"},"comment":{"url":"https://api.github.com/repos/ReactiveX/RxJava/issues/comments/83541029","html_url":"https://github.com/ReactiveX/RxJava/pull/2641#issuecomment-83541029","issue_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/2641","id":83541029,"user":{"login":"mttkay","id":102802,"avatar_url":"https://avatars.githubusercontent.com/u/102802?v=3","gravatar_id":"","url":"https://api.github.com/users/mttkay","html_url":"https://github.com/mttkay","followers_url":"https://api.github.com/users/mttkay/followers","following_url":"https://api.github.com/users/mttkay/following{/other_user}","gists_url":"https://api.github.com/users/mttkay/gists{/gist_id}","starred_url":"https://api.github.com/users/mttkay/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mttkay/subscriptions","organizations_url":"https://api.github.com/users/mttkay/orgs","repos_url":"https://api.github.com/users/mttkay/repos","events_url":"https://api.github.com/users/mttkay/events{/privacy}","received_events_url":"https://api.github.com/users/mttkay/received_events","type":"User","site_admin":false},"created_at":"2015-03-19T12:46:04Z","updated_at":"2015-03-19T12:46:04Z","body":"Deferred might get confused with defer/OnSubscribeDefer though?\n\nOn Thu, Mar 19, 2015 at 11:12 AM, Sébastien Deleuze <\nnotifications@github.com> wrote:\n\n> Hi @benjchristensen <https://github.com/benjchristensen>,\n>\n> I fully agree with the need to have a scalar observable type, especially\n> when exposing such type in application public API.\n>\n> The main thing that confuses me in your implementation is this embedded\n> Task.Promise type. I understand this is done differently in .NET since\n> Task already exists, but other platforms I would just expect a single\n> type that I would expose in my API instead of Observable for scalar\n> result.\n>\n> About the naming:\n>\n>    - I agree that Task is confusing\n>    - ScalarObservable is the most meaningful name, but I agree with\n>    @tomlokhorst <https://github.com/tomlokhorst>, people will expect it\n>    to extends Observable and that not the case\n>    - What about Deferred?\n>\n> Thanks in advance for your feedback.\n>\n> —\n> Reply to this email directly or view it on GitHub\n> <https://github.com/ReactiveX/RxJava/pull/2641#issuecomment-83479377>.\n>\n\n\n\n-- \n\nMatthias Käppler\n\n*Engineer*\n\n\nTwitter: https://twitter.com/mttkay\n\nSkype: matthias-sc\n\n\n\nSoundCloud Ltd. | Rheinsberger Str. 76/77, 10115 Berlin, Germany | +49\n(0)172 2345679\n\nManaging Director: Alexander Ljung | Incorporated in England & Wales\nwith Company\nNo. 6343600 | Local Branch Office | AG Charlottenburg  | HRB 110657B\n\n\n\nCapture and share your music & audio on SoundCloud\n<http://soundcloud.com/creators>\n"}}