gRPC 와 android java를 활용한 채팅 어플 만들기

프로토콜 정의 

service ChatService {
// Sends a greeting
rpc SayHello (HelloRequest) returns (HelloReply) {}

rpc CreateRoom(CreateRoomRequest) returns (stream ChatMessage) {}
rpc JoinRoom(JoinRoomRequest) returns (stream ChatMessage) {}
rpc SendChat(SendChatRequest) returns (SendChatReply) {}
rpc GetChatRooms(GetChatRoomsRequest) returns (GetChatRoomsReply) {}
rpc LeaveRoom(LeaveRoomRequest) returns (LeaveRoomReply) {}
}

대략 채팅에 필요한 RPC를 정의한다. 

 

1) CreateRoom에서 요청(CreateRoomRequest)은 일반 패킷이고, 응답(ChatMessage)은 스트립으로 정의했다 이유는 방생성을 하면 해당 방에서 수신되는 메시지를 계속적으로 받기 위해서 이다.

 

2) SendChat으로 채팅 방에 메시지를 송신하면 서버는 해당 방에 입장한 유저에 브로드케스팅하게 된다.

 

package com.stfalcon.chatkit.sample.features.demo.def;

import android.app.Activity;
import android.content.Context;
import android.content.Intent;
import android.os.AsyncTask;
import android.os.Bundle;
import android.text.TextUtils;
import android.util.Log;

import com.stfalcon.chatkit.messages.MessageInput;
import com.stfalcon.chatkit.messages.MessagesList;
import com.stfalcon.chatkit.messages.MessagesListAdapter;
import com.stfalcon.chatkit.sample.R;
import com.stfalcon.chatkit.sample.common.data.fixtures.MessagesFixtures;
import com.stfalcon.chatkit.sample.features.demo.DemoMessagesActivity;
import com.stfalcon.chatkit.sample.utils.AppUtils;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.lang.ref.WeakReference;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.gamechat.ChatMessage;
import io.grpc.gamechat.ChatServiceGrpc;
import io.grpc.gamechat.CreateRoomRequest;
import io.grpc.gamechat.HelloReply;
import io.grpc.gamechat.HelloRequest;
import io.grpc.gamechat.SendChatReply;
import io.grpc.gamechat.SendChatRequest;
import io.grpc.stub.StreamObserver;


public class DefaultMessagesActivity extends DemoMessagesActivity
        implements MessageInput.InputListener,
        MessageInput.AttachmentsListener,
        MessageInput.TypingListener {

    public static void open(Context context) {
        context.startActivity(new Intent(context, DefaultMessagesActivity.class));
    }

    private MessagesList messagesList;


    private ManagedChannel channel;
    private ChatServiceGrpc.ChatServiceStub asyncStub;
    private String roomId;


    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_default_messages);

        this.messagesList = findViewById(R.id.messagesList);
        initAdapter();

        MessageInput input = findViewById(R.id.input);
        input.setInputListener(this);
        input.setTypingListener(this);
        input.setAttachmentsListener(this);


        String host = "10.0.2.2";
        int port = 50051;
        try {
            channel = ManagedChannelBuilder
                    .forAddress(host, port)
                    .usePlaintext()
                    .keepAliveTime(30, TimeUnit.SECONDS)
                    .keepAliveWithoutCalls(true)
                    .build();
            asyncStub = ChatServiceGrpc.newStub(channel);
            createRoom();

        } catch (Exception e) {
            StringWriter sw = new StringWriter();
            PrintWriter pw = new PrintWriter(sw);
            e.printStackTrace(pw);
            pw.flush();
            Log.d("error", sw.toString());
        }
    }

    @Override
    public boolean onSubmit(CharSequence input) {
//        super.messagesAdapter.addToStart(
//                MessagesFixtures.getTextMessage(input.toString()), true);
        sendMessage(input.toString());
        return true;
    }

    @Override
    public void onAddAttachments() {
        super.messagesAdapter.addToStart(
                MessagesFixtures.getImageMessage(), true);
    }

    private void initAdapter() {
        super.messagesAdapter = new MessagesListAdapter<>(super.senderId, super.imageLoader);
        super.messagesAdapter.enableSelectionMode(this);
        super.messagesAdapter.setLoadMoreListener(this);
        super.messagesAdapter.registerViewClickListener(R.id.messageUserAvatar,
                (view, message) -> AppUtils.showToast(DefaultMessagesActivity.this,
                        message.getUser().getName() + " avatar click",
                        false));
        this.messagesList.setAdapter(super.messagesAdapter);
    }

    @Override
    public void onStartTyping() {
        Log.v("Typing listener", getString(R.string.start_typing_status));
    }

    @Override
    public void onStopTyping() {
        Log.v("Typing listener", getString(R.string.stop_typing_status));
    }


    public void recvMessage(String msg)
    {
        super.messagesAdapter.addToStart(
                MessagesFixtures.getTextMessage(msg), true);
    }

    public void sendMessage(String msg) {
//        ((InputMethodManager) getSystemService(Context.INPUT_METHOD_SERVICE))
//                .hideSoftInputFromWindow(hostEdit.getWindowToken(), 0);
//        sendButton.setEnabled(false);
//        resultText.setText("");
        new GrpcTask(this)
                .execute(
                        "10.0.2.2",
                        msg,
                        "50051",
                        roomId
                );
    }

    private static class GrpcTask extends AsyncTask<String, Void, String> {
        private final WeakReference<Activity> activityReference;
        private ManagedChannel channel;

        private GrpcTask(Activity activity) {
            this.activityReference = new WeakReference<Activity>(activity);
        }

        @Override
        protected String doInBackground(String... params) {
            String host = params[0];
            String message = params[1];
            String portStr = params[2];
            String roomId = params[3];
            int port = TextUtils.isEmpty(portStr) ? 0 : Integer.valueOf(portStr);
            try {
                channel = ManagedChannelBuilder.forAddress(host, port).usePlaintext().build();
                ChatServiceGrpc.ChatServiceBlockingStub stub = ChatServiceGrpc.newBlockingStub(channel);
                SendChatRequest request = SendChatRequest.newBuilder().setMessage(message).setRoomId(roomId).build();
                SendChatReply reply = stub.sendChat(request);
                Log.d("SendChatReply", reply.getCode().toString());

                return message;
            } catch (Exception e) {
                StringWriter sw = new StringWriter();
                PrintWriter pw = new PrintWriter(sw);
                e.printStackTrace(pw);
                pw.flush();
                return String.format("Failed... : %n%s", sw);
            }
        }

        @Override
        protected void onPostExecute(String result) {
            try {
                channel.shutdown().awaitTermination(1, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            Activity activity = activityReference.get();
            if (activity == null) {
                return;
            }
//            TextView resultText = (TextView) activity.findViewById(R.id.grpc_response_text);
//            Button sendButton = (Button) activity.findViewById(R.id.send_button);
//            resultText.setText(result);
//            sendButton.setEnabled(true);

            Log.d("Game", result);
            //((DefaultMessagesActivity)activity).recvMessage(result);
        }
    }



    private void createRoom()
            throws InterruptedException, RuntimeException {

        CreateRoomRequest request = CreateRoomRequest.newBuilder()
                .setRoomName("test")
                .setSenderName("username")
                .build();

        StreamObserver<ChatMessage> response = new StreamObserver<ChatMessage> (){

            @Override
            public void onNext(ChatMessage value) {
                Log.d("recv msg", value.getMessage());
                roomId = value.getRoomId();

                runOnUiThread(new Runnable() {
                    @Override
                    public void run() {
                        recvMessage(value.getMessage());
                    }
                });
            }

            @Override
            public void onError(Throwable t)
            {
            }

            @Override
            public void onCompleted()
            {
            }
        };

        asyncStub.createRoom(request, response);
    }



}

채팅방을 개설하면 비동기(asyncStub)로 메시지를 수신 받게 되는데, 이때 onNext가 호출되고 메시지가 온 것을 확인할 수 있다. 다만 해당 스레드와 ui 를 처리하는 스레드가 달라서 ui 스레드로 전달해줘야 정상적으로 처리가 된다.

이 부분을 android 개발 초보인 저로서는 원인을 알지 못해서 많이 헤매었다. 

 recvMessage 직접 호출시 서버쪽에서 에러가 발생하는데  already finished라는 예외가 발생하여  혹시 asyncStub 의 요청이 무효화 되는것 아닌지 의심하고 삽질을 했다.