我已经完成了几个教程，比如这个和这个关于使用Rxjava作为EventBus的教程。然而，它们并不能解决我的问题。
我正在使用我自己的自定义nodejs服务器和socket.io制作聊天应用程序。授权是通过改造完成的(使用gson)。即使应用程序被关闭，用户也应该收到消息，因此连接到socket。io是通过一个长期服务来维护的，如果信息显示给用户，它将把信息转发给活动，否则它只是显示一个通知并写入SQL db (TODO)。我的Service接收socket.io的呼叫并解析负载。当这种情况发生时，它需要将消息与接收到的有效负载一起发送给活动。
MyEventBus.class
public class MyRxBus {
private static MyRxBus instance;
private PublishSubject<Object> 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<Object> getEvents() {
return subject;
}
}
SocketService类
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)
}
MainActivity.class
public class MainActivity extends VesicaActivity {
private Socket mSocketOne;
private Gson gson = new Gson();
private List<ChatMessage> mMessages = new ArrayList<ChatMessage>();
private List<String> mListOfUsers = new ArrayList<String>();
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<Object>() {
@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());
}
}
当我运行这段代码时，我得到以下错误rx.exceptions.OnErrorNotImplementedException: Can't create handler inside thread that has not called Looper.prepare()
指向at com.example.varun.vesica.eventbus.MyRxBus.setEvent(MyRxBus.java:26)
我知道可观察对象和观察者可以设置为发出/观察不同的线程，但是经过大量的试验和错误，我已经放弃了，需要一些帮助。
任何帮助/链接将不胜感激。谢谢!
一直保持连接会耗尽用户的电量。当设备被屏蔽或应用被最小化时，你应该考虑使用推送通知。当你在Activity中订阅Observable时，保存Subscription并在onPause中关闭它，否则会因为设备旋转而导致连接延迟。考虑使用GreenRobot事件总线而不是重新发明轮子，它是连接服务和活动的非常方便的库:http://www.andreas-schrade.de/2015/11/28/android-how-to-use-the-greenrobot-eventbus/
您试图在调用setEvent()/setString()的任意线程上调用adapter.notifyDataSetChanged()和Toast.makeText()。当你以更新Android UI为目标订阅getEvents()可观察对象时，你需要在主线程调度程序上观察:
MyRxBus.instanceOf()
.getEvents()
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<Object>() {
@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();
}
});