RxJava as an Eventbus between service and activity
<p>I have gone through several tutorials such as <a href="https://lorentzos.com/rxjava-as-event-bus-the-right-way-10a36bdd49ba#.k2tzbjabu" rel="nofollow">this</a> and <a href="http://blog.kaush.co/2014/12/24/implementing-an-event-bus-with-rxjava-rxbus/" rel="nofollow">this</a> on the usage of <code>Rxjava</code> as an <code>EventBus</code>. However, they do not solve my problem. </p>

<p>I am making a chat app using my own custom nodejs server and socket.io. Authorization is done via retrofit (with gson). The user should receive messages even if the app is switched off, hence the connection to socket.io is maintained through a perennial service, which will relay information to the activity IF it is displayed to the user, else it simply shows a notification and writes to SQL db (TODO). My <code>Service</code> receives a call from <code>socket.io</code> and parses the payload. When that happens, it needs to send a message to the activity along with the payload received.</p>

<p>MyEventBus.class</p>

<pre><code>public class MyRxBus {

    private static MyRxBus instance;

    private PublishSubject&lt;Object&gt; subject = PublishSubject.create();

    public static MyRxBus instanceOf() {
        if (instance == null) {
            instance = new MyRxBus();
        }
        return instance;
    }

    /**
     * Pass any event down to event listeners.
     */
    public void setString(Object object) {
        subject.onNext(object);
    }

    /**
     * Subscribe to this Observable. On event, do something 
     * e.g. replace a fragment
     */
    public Observable&lt;Object&gt; getEvents() {
        return subject;
    }
}
</code></pre>

<p>SocketService Class</p>

<pre><code>public class SocketService extends Service {
    private Socket mSocket;
    public SocketService() {
    }

    @Override
    public void onCreate() {
        super.onCreate();
        Log.d("SERVICE ","SOCKET SERVICE WAS CALLED");
        mSocket=SocketSingleton.getSocket("8080");
        mSocket.on(Socket.EVENT_CONNECT,onConnect);
        mSocket.on("new_message",onNewMessage);
        mSocket.connect();
    }


    private Emitter.Listener onNewMessage = new Emitter.Listener() {
        @Override
        public void call(Object... args) {
            Log.d("Received from socket ",args[0].toString());
            try {
                final JSONObject object = new JSONObject((String) args[0]);
                //Send this object to all subscribers
                //TODO: add rxjava eventbus to 1)add info to db, 2)update activity IF it is displayed
                MyRxBus.instanceOf()
                    .setEvent(object);

            } catch (JSONException e) {
                e.printStackTrace();
            }
        }
    };
  //some other functions (not related at all)

}
</code></pre>

<p>MainActivity.class</p>

<pre><code>   public class MainActivity extends VesicaActivity {
    private Socket mSocketOne;
    private Gson gson = new Gson();
    private List&lt;ChatMessage&gt; mMessages = new ArrayList&lt;ChatMessage&gt;();
    private List&lt;String&gt; mListOfUsers = new ArrayList&lt;String&gt;();
    Boolean isNodeOne;
    String username;
    private JSONArray arrayOfUsers=null;
    private RecyclerView mRecyclerView;
    private RecyclerView.Adapter mAdapter;
    private RecyclerView.LayoutManager mLayoutManager;
    private Boolean _isConnected=false;
    private String socketIdtoSend;

    @BindView(R.id.et_text)EditText etText;
    @BindView(R.id.tv_user_list)TextView tvListUsers;
    @BindView(R.id.tv_status)TextView tvStatus;
    @BindView(R.id.toolbar)Toolbar toolbar;
    @BindView(R.id.et_self_destruct_time)EditText etSelfDestructTimer;
    @BindView(R.id.cb_self_destruct)CheckBox cbSelfDestruct;


    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        ButterKnife.bind(this);
        setSupportActionBar(toolbar);
        if (getSupportActionBar()!=null){
            getSupportActionBar().setDisplayHomeAsUpEnabled(true);
        }
        tvStatus.setText("checking connection...");
        username = Constants.getUsername();
        Integer portNumber = Constants.getPortNumber();
        socketIdtoSend = getIntent().getStringExtra("socketId");
        mMessages.add(createMessage("Sample Name","Sample Text",0));
        mRecyclerView = (RecyclerView) findViewById(R.id.recycler_view);
        mLayoutManager = new LinearLayoutManager(this);
        mRecyclerView.setLayoutManager(mLayoutManager);
        mAdapter = new ChatAdapter(mMessages, username);
        mRecyclerView.setAdapter(mAdapter);


        //-----------------------------------------------------------------

        MyRxBus.instanceOf()
                .getEvents()
                .subscribe(new Action1&lt;Object&gt;() {
                    @Override
                    public void call(Object o) {
                       final JSONObject object = new JSONObject((String) o)
                       ChatMessage message = createMessage(object.getString("user")
                               ,Encryption.Decrypt(object.getString("message"))
                               ,object.getInt("selfDestructTime"));
                       mMessages.add(message);
                       mAdapter.notifyDataSetChanged();
                        Toast.makeText(MainActivity.this," new message received ", Toast.LENGTH_SHORT).show();
                    }
                });

    }


    public ChatMessage createMessage(String user, String message, Integer time){
        return new ChatMessage.ChatMessageBuilder()
                .message(message)
                .user(user)
                .type("newMessage")
                .selfDestructTime(time)
                .sendToSocketId(socketIdtoSend)
                .build();
    }


    public void showDialogBoxToClearHistory(){
        new AlertDialog.Builder(MainActivity.this)
                .setTitle("Wipe Chat History")
                .setMessage("This will delete the entire chat history of this thread. It is irreversible. Do you want to continue?")
                .setPositiveButton(android.R.string.yes, new DialogInterface.OnClickListener() {
                    public void onClick(DialogInterface dialog, int which) {
                        // continue with delete
                        mMessages.clear();
                        mAdapter.notifyDataSetChanged();
                        Toast.makeText(MainActivity.this, "Cleared entire history", Toast.LENGTH_LONG).show();
                    }
                })
                .setNegativeButton("Skip", new DialogInterface.OnClickListener() {
                    public void onClick(DialogInterface dialog, int which) {
                       try{
                           dialog.dismiss();
                       }catch (Exception e){e.printStackTrace();}
                    }
                })
                .setCancelable(false)
                .show();
    }

    private void ScrollToBottom(){
        mRecyclerView.smoothScrollToPosition(mAdapter.getItemCount());
    }
}
</code></pre>

<p>When I run this code, I get the following error
 <code>rx.exceptions.OnErrorNotImplementedException: Can't create handler inside thread that has not called Looper.prepare()</code></p>

<p>that points to
<code>at com.example.varun.vesica.eventbus.MyRxBus.setEvent(MyRxBus.java:26)</code></p>

<p>I am aware that observables and observers can be set to emit/ observe of different threads, however after lot of trial and error, I have given up and need some help.</p>

<p>Any help/ links will be appreciated. Thanks!</p>