{"action":"opened","issue":{"url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3600","labels_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3600/labels{/name}","comments_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3600/comments","events_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3600/events","html_url":"https://github.com/ReactiveX/RxJava/issues/3600","id":124978020,"number":3600,"title":"Call onNext from Main thread dont repect subscribeOn when flatMap or switchMap","user":{"login":"danielgomezrico","id":76348,"avatar_url":"https://avatars.githubusercontent.com/u/76348?v=3","gravatar_id":"","url":"https://api.github.com/users/danielgomezrico","html_url":"https://github.com/danielgomezrico","followers_url":"https://api.github.com/users/danielgomezrico/followers","following_url":"https://api.github.com/users/danielgomezrico/following{/other_user}","gists_url":"https://api.github.com/users/danielgomezrico/gists{/gist_id}","starred_url":"https://api.github.com/users/danielgomezrico/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/danielgomezrico/subscriptions","organizations_url":"https://api.github.com/users/danielgomezrico/orgs","repos_url":"https://api.github.com/users/danielgomezrico/repos","events_url":"https://api.github.com/users/danielgomezrico/events{/privacy}","received_events_url":"https://api.github.com/users/danielgomezrico/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":0,"created_at":"2016-01-05T14:32:37Z","updated_at":"2016-01-05T14:32:37Z","closed_at":null,"body":"Im doing a wrapper for the facebook sdk and trying to do it with rx java but Im getting an issue with the actual thread for the observables created in my wrapper. I want to get the user object after the user log in with fb.\r\n\r\n```\r\npublic class LoginPresenter extends BasePresenter<LoginView> {\r\n\r\npublic void startFacebookLogin() {\r\n    if (!mNetworkChecker.isNetworkConnectionAvailable()) {\r\n      mView.showNoConnectionError();\r\n      return;\r\n    }\r\n\r\n    Subscription subscription = mFacebookRepository.startLoginFlow()\r\n        .flatMap(loginResult -> mFacebookRepository.getUser())\r\n        .compose(Transformer.applyIoSchedulers())\r\n        .subscribe(user -> mView.startSignUpView(user), this::manageErrorFlow);\r\n\r\n    mCompositeSubscription.add(subscription);\r\n  }\r\n}\r\n```\r\n\r\n```\r\npublic class FacebookRepository {\r\n\r\n  private WeakReference<Activity> mActivityWeakReference;\r\n  @Nullable private CallbackManager callbackManager;\r\n\r\n  public FacebookRepository(String facebookPublicKey, Activity activity) {\r\n    if (!FacebookSdk.isInitialized()) {\r\n      FacebookSdk.sdkInitialize(com.barista_v.wapa.api.Api.getContext());\r\n    }\r\n\r\n    if (com.barista_v.wapa.api.Api.getLogLevel() > com.barista_v.wapa.api.Api.LOG_LEVEL_BASIC) {\r\n      FacebookSdk.addLoggingBehavior(LoggingBehavior.REQUESTS);\r\n    }\r\n\r\n    mActivityWeakReference = new WeakReference<>(activity);\r\n\r\n    FacebookSdk.setApplicationId(facebookPublicKey);\r\n  }\r\n\r\n  public Observable<LoginResult> startLoginFlow() {\r\n    return Observable.create(subscriber -> {\r\n      Activity activity = mActivityWeakReference.get();\r\n      if (activity != null) {\r\n        LoginManager loginManager = LoginManager.getInstance();\r\n        callbackManager = CallbackManager.Factory.create();\r\n\r\n        loginManager.logInWithReadPermissions(activity, getPermissions());\r\n        loginManager.registerCallback(callbackManager, new FacebookCallback<LoginResult>() {\r\n          @Override\r\n          public void onSuccess(LoginResult loginResult) {\r\n            subscriber.onNext(loginResult);\r\n            subscriber.onCompleted();\r\n          }\r\n\r\n          @Override\r\n          public void onCancel() {\r\n            subscriber.onError(\r\n                new FacebookLoginCancelException(\"Login canceled.\", \"Facebook Error\", null));\r\n          }\r\n\r\n          @Override\r\n          public void onError(FacebookException exception) {\r\n            subscriber.onError(new FacebookLoginException(exception));\r\n          }\r\n        });\r\n      } else {\r\n        Timber.w(\"The activity reference is null for FacebookRepository.startLoginFlow()\");\r\n        subscriber.onError(\r\n            new FacebookLoginException(\"Something happened internally.\", \"Facebook Error\", null));\r\n      }\r\n    });\r\n  }\r\n\r\n  public Observable<User> getUser() {\r\n    return Observable.defer(() -> {\r\n      AccessToken accessToken = AccessToken.getCurrentAccessToken();\r\n      String userId = accessToken.getUserId();\r\n      Bundle params = new Bundle();\r\n      params.putString(\"fields\", \"name,email\");\r\n      String graphPath = \"/\" + userId;\r\n      GraphRequest graphRequest = new GraphRequest(accessToken, graphPath, params, HttpMethod.GET);\r\n\r\n      JSONObject json;\r\n      try {\r\n        GraphResponse result = graphRequest.executeAndWait();\r\n        json = result.getJSONObject();\r\n      } catch (FacebookException e) {\r\n        return Observable.error(e);\r\n      }\r\n\r\n      String name, email;\r\n\r\n      try {\r\n        name = json.getString(\"name\");\r\n      } catch (JSONException | JsonSyntaxException | NullPointerException e) {\r\n        return Observable.error(e);\r\n      }\r\n\r\n      try {\r\n        email = json.getString(\"email\");\r\n      } catch (JSONException | JsonSyntaxException | NullPointerException e) {\r\n        com.barista_v.wapa.api.error_handling.ApiException exception =\r\n            new UserInfoMissingException(\r\n                \"Email access permission is not granted but it is required.\",\r\n                \"Invalid Facebook Permissions\", e);\r\n        return Observable.error(exception);\r\n      }\r\n\r\n      String photoUrl =\r\n          String.format(\"https://graph.facebook.com/%s/picture?type=large\", userId);\r\n\r\n      return Observable.just(new User(name, email, photoUrl));\r\n    });\r\n  }\r\n\r\n  public void onActivityResult(int requestCode, int resultCode, Intent data) {\r\n    if (callbackManager != null) {\r\n      callbackManager.onActivityResult(requestCode, resultCode, data);\r\n    }\r\n  }\r\n}\r\n\r\npublic class Transformer {\r\n  public static <T> Observable.Transformer<T, T> applyIoSchedulers() {\r\n    return observable -> observable.subscribeOn(Schedulers.io())\r\n        .observeOn(AndroidSchedulers.mainThread());\r\n  }\r\n}\r\n```\r\n\r\nThe problem is that the code that execute the `mFacebookRepository.getUser()` is in the main thread.\r\nShould not be the IO thread?\r\n\r\nI create the observable and call onNext when `FacebookCallback` call onSuccess function and it is called in the main thread, so subscriber.onNext is called on the main thread.\r\n\r\nIs this a bug or something Im doing wrong?"}}