{"action":"created","issue":{"url":"https://api.github.com/repos/Reactive-Extensions/Rx.NET/issues/304","repository_url":"https://api.github.com/repos/Reactive-Extensions/Rx.NET","labels_url":"https://api.github.com/repos/Reactive-Extensions/Rx.NET/issues/304/labels{/name}","comments_url":"https://api.github.com/repos/Reactive-Extensions/Rx.NET/issues/304/comments","events_url":"https://api.github.com/repos/Reactive-Extensions/Rx.NET/issues/304/events","html_url":"https://github.com/Reactive-Extensions/Rx.NET/issues/304","id":194034405,"number":304,"title":"Strange behavior with Observable.Create<T>(async subscription)","user":{"login":"softeering","id":6609594,"avatar_url":"https://avatars.githubusercontent.com/u/6609594?v=3","gravatar_id":"","url":"https://api.github.com/users/softeering","html_url":"https://github.com/softeering","followers_url":"https://api.github.com/users/softeering/followers","following_url":"https://api.github.com/users/softeering/following{/other_user}","gists_url":"https://api.github.com/users/softeering/gists{/gist_id}","starred_url":"https://api.github.com/users/softeering/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/softeering/subscriptions","organizations_url":"https://api.github.com/users/softeering/orgs","repos_url":"https://api.github.com/users/softeering/repos","events_url":"https://api.github.com/users/softeering/events{/privacy}","received_events_url":"https://api.github.com/users/softeering/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":7,"created_at":"2016-12-07T12:15:45Z","updated_at":"2016-12-08T11:59:13Z","closed_at":null,"body":"Hi there,\r\n\r\nI'm totally new to Rx and there is something I don't understand when trying to create an observable with an ASYNC subscription.\r\nLet's assume I have a console app and I run the code below:\r\n```\r\nclass Program\r\n{\r\n\tstatic void Main(string[] args)\r\n\t{\r\n\t\tnew Program().Run();\r\n\t\tConsole.WriteLine(\"FULL COMPLETE\");\r\n\t\tConsole.ReadLine();\r\n\t}\r\n\r\n\tpublic void Run()\r\n\t{\r\n\t\tConsole.WriteLine(\"BEFORE SYNC SUB\");\r\n\t\tthis.SyncObservable().Subscribe(i =>\r\n\t\t{\r\n\t\t\tConsole.WriteLine($\"SYNC {i}\");\r\n\t\t});\r\n\t\tConsole.WriteLine(\"AFTER SYNC SUB\");\r\n\r\n\t\tConsole.WriteLine(\"BEFORE ASYNC SUB\");\r\n\t\tthis.AsyncObservable().Subscribe(i =>\r\n\t\t{\r\n\t\t\tConsole.WriteLine($\"ASYNC {i}\");\r\n\t\t});\r\n\t\tConsole.WriteLine(\"AFTER ASYNC SUB\");\r\n\t}\r\n\r\n\tpublic IObservable<int> SyncObservable()\r\n\t{\r\n\t\treturn Observable.Range(0, 10);\r\n\t}\r\n\r\n\tpublic IObservable<int> AsyncObservable()\r\n\t{\r\n\t\treturn Observable.Create<int>(async (observer, cancellationToken) =>\r\n\t\t{\r\n\t\t\t// let's say this lambda connects to a SQL db and asynchronously executes a reader\r\n\t\t\tfor (int i = 0; i < 10; i++)\r\n\t\t\t{\r\n\t\t\t\tawait Task.Delay(50);\r\n\t\t\t\tobserver.OnNext(i);\r\n\t\t\t}\r\n\t\t\tobserver.OnCompleted();\r\n\t\t});\r\n\t}\r\n}\r\n```\r\nThe output is the following:\r\n```\r\nBEFORE SYNC SUB\r\nSYNC 0\r\nSYNC 1\r\nSYNC 2\r\nSYNC 3\r\nSYNC 4\r\nSYNC 5\r\nSYNC 6\r\nSYNC 7\r\nSYNC 8\r\nSYNC 9\r\nAFTER SYNC SUB\r\nBEFORE ASYNC SUB\r\nAFTER ASYNC SUB\r\nFULL COMPLETE\r\nASYNC 0\r\nASYNC 1\r\nASYNC 2\r\nASYNC 3\r\nASYNC 4\r\nASYNC 5\r\nASYNC 6\r\nASYNC 7\r\nASYNC 8\r\nASYNC 9\r\n```\r\nBut I would have expected this:\r\n```\r\nBEFORE SYNC SUB\r\nSYNC 0\r\nSYNC 1\r\nSYNC 2\r\nSYNC 3\r\nSYNC 4\r\nSYNC 5\r\nSYNC 6\r\nSYNC 7\r\nSYNC 8\r\nSYNC 9\r\nAFTER SYNC SUB\r\nBEFORE ASYNC SUB\r\nASYNC 0\r\nASYNC 1\r\nASYNC 2\r\nASYNC 3\r\nASYNC 4\r\nASYNC 5\r\nASYNC 6\r\nASYNC 7\r\nASYNC 8\r\nASYNC 9\r\nAFTER ASYNC SUB\r\nFULL COMPLETE\r\n```\r\n\r\nI understand that the lambda passed to Create<int> is asynchronous but why is .Subscribe not waiting for the .OnCompleted() method to be called to go back to the normal flow?\r\n\r\nI don't want Subscribe() to be executed \"in the background\", I just need to create an Observable on top of an ASYNC IO operation (read from database for example). I'd like the Subscribe method to behave as the one called on the Observable created via Observable.Range(...)\r\n\r\nOther question: is there a way to call some async code in the .Subscribe(..) lambda and await?\r\n\r\nThanks in advance for your help\r\n\r\n-David\r\n\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/Reactive-Extensions/Rx.NET/issues/comments/265723989","html_url":"https://github.com/Reactive-Extensions/Rx.NET/issues/304#issuecomment-265723989","issue_url":"https://api.github.com/repos/Reactive-Extensions/Rx.NET/issues/304","id":265723989,"user":{"login":"softeering","id":6609594,"avatar_url":"https://avatars.githubusercontent.com/u/6609594?v=3","gravatar_id":"","url":"https://api.github.com/users/softeering","html_url":"https://github.com/softeering","followers_url":"https://api.github.com/users/softeering/followers","following_url":"https://api.github.com/users/softeering/following{/other_user}","gists_url":"https://api.github.com/users/softeering/gists{/gist_id}","starred_url":"https://api.github.com/users/softeering/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/softeering/subscriptions","organizations_url":"https://api.github.com/users/softeering/orgs","repos_url":"https://api.github.com/users/softeering/repos","events_url":"https://api.github.com/users/softeering/events{/privacy}","received_events_url":"https://api.github.com/users/softeering/received_events","type":"User","site_admin":false},"created_at":"2016-12-08T11:59:13Z","updated_at":"2016-12-08T11:59:13Z","body":"OK let's take this code\r\n```\r\nusing System;\r\nusing System.Data.SqlClient;\r\nusing System.Reactive.Disposables;\r\nusing System.Reactive.Linq;\r\n\r\nnamespace ConsoleApplication2\r\n{\r\n\tclass Program\r\n\t{\r\n\t\tprivate const string ConnectionString = \"server=localhost; database=TEST; integrated security=true\";\r\n\t\tprivate const string Query = \"select top 50 ID from TESTKEY\";\r\n\r\n\t\tstatic void Main(string[] args)\r\n\t\t{\r\n\t\t\tnew Program().Run();\r\n\t\t\tConsole.WriteLine(\"FULL COMPLETE\");\r\n\t\t\tConsole.ReadLine();\r\n\t\t}\r\n\r\n\t\tpublic void Run()\r\n\t\t{\r\n\t\t\tConsole.WriteLine(\"BEFORE SYNC READ\");\r\n\t\t\tthis.SyncRead().Subscribe(i =>\r\n\t\t\t{\r\n\t\t\t\tConsole.WriteLine($\"SYNC {i}\");\r\n\t\t\t});\r\n\t\t\tConsole.WriteLine(\"AFTER SYNC READ\");\r\n\r\n\t\t\tConsole.WriteLine(\"BEFORE ASYNC READ\");\r\n\t\t\tthis.AsyncRead().Subscribe(i =>\r\n\t\t\t{\r\n\t\t\t\tConsole.WriteLine($\"ASYNC {i}\");\r\n\t\t\t});\r\n\t\t\tConsole.WriteLine(\"AFTER ASYNC READ\");\r\n\t\t}\r\n\r\n\t\tpublic IObservable<int> SyncRead()\r\n\t\t{\r\n\t\t\treturn Observable.Create<int>(observer =>\r\n\t\t\t{\r\n\t\t\t\tusing (var connection = new SqlConnection(ConnectionString))\r\n\t\t\t\t{\r\n\t\t\t\t\tconnection.Open();\r\n\r\n\t\t\t\t\tusing (var command = connection.CreateCommand())\r\n\t\t\t\t\t{\r\n\t\t\t\t\t\tcommand.CommandText = Query;\r\n\t\t\t\t\t\tusing (var reader = command.ExecuteReader())\r\n\t\t\t\t\t\t{\r\n\t\t\t\t\t\t\twhile (reader.Read())\r\n\t\t\t\t\t\t\t{\r\n\t\t\t\t\t\t\t\tobserver.OnNext(reader.GetInt32(0));\r\n\t\t\t\t\t\t\t}\r\n\r\n\t\t\t\t\t\t\tobserver.OnCompleted();\r\n\t\t\t\t\t\t}\r\n\t\t\t\t\t}\r\n\t\t\t\t}\r\n\r\n\t\t\t\treturn Disposable.Empty;\r\n\t\t\t});\r\n\t\t}\r\n\r\n\t\tpublic IObservable<int> AsyncRead()\r\n\t\t{\r\n\t\t\treturn Observable.Create<int>(async (observer, cancellationToken) =>\r\n\t\t\t{\r\n\t\t\t\tusing (var connection = new SqlConnection(ConnectionString))\r\n\t\t\t\t{\r\n\t\t\t\t\tawait connection.OpenAsync().ConfigureAwait(false);\r\n\r\n\t\t\t\t\tusing (var command = connection.CreateCommand())\r\n\t\t\t\t\t{\r\n\t\t\t\t\t\tcommand.CommandText = Query;\r\n\t\t\t\t\t\tusing (var reader = await command.ExecuteReaderAsync().ConfigureAwait(false))\r\n\t\t\t\t\t\t{\r\n\t\t\t\t\t\t\twhile (await reader.ReadAsync().ConfigureAwait(false))\r\n\t\t\t\t\t\t\t{\r\n\t\t\t\t\t\t\t\tobserver.OnNext(reader.GetInt32(0));\r\n\t\t\t\t\t\t\t}\r\n\r\n\t\t\t\t\t\t\tobserver.OnCompleted();\r\n\t\t\t\t\t\t}\r\n\t\t\t\t\t}\r\n\t\t\t\t}\r\n\t\t\t});\r\n\t\t}\r\n\t}\r\n}\r\n```\r\nAnd here is the output\r\n```\r\nBEFORE SYNC READ\r\nSYNC 1\r\nSYNC 1\r\nSYNC 1\r\nSYNC 1\r\nSYNC 1\r\n...\r\nSYNC 1\r\nSYNC 1\r\nSYNC 1\r\nSYNC 1\r\nSYNC 1\r\nAFTER SYNC READ\r\nBEFORE ASYNC READ\r\nAFTER ASYNC READ\r\nFULL COMPLETE\r\nASYNC 1\r\nASYNC 1\r\nASYNC 1\r\nASYNC 1\r\nASYNC 1\r\n...\r\nASYNC 1\r\nASYNC 1\r\nASYNC 1\r\nASYNC 1\r\nASYNC 1\r\n```\r\n\r\nI don't get why they behave differently in terms of flow.\r\nI don't want the second one to be executed in the background, I'm just using ASYNC / AWAIT IO operations.\r\nWhy is the underlying Task passed to Observable.Create not awaited until the .OnCompleted call? (same as the synchronous code does)\r\n\r\nThanks\r\n"}}
{"actor":{"display_login":"softeering"}}