Playframework付属のサンプル'websocket-chat'を見てみる(3/4)

[前回](http://qiita.com/yyyske/items/0d00e5e352b7078a501e)の続き。今回はサーバ側のWebSocketを見てみる。

#Application.java
```java:Application.java
public static WebSocket<JsonNode> chat(final String username) {
	
    return new WebSocket<JsonNode>() {
        // Called when the Websocket Handshake is done.
        public void onReady(WebSocket.In<JsonNode> in, WebSocket.Out<JsonNode> out){
            
            // Join the chat room.
            try { 
                ChatRoom.join(username, in, out);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
    };
}
```
クライアントにWebSocketを返すところで、無名クラスを定義している。このクラスには`onReady`というメソッドがあるが、これはハンドシェイクが完了した時点で実行されるとのこと。おそらく最初に一回呼ばれるのだろうね。引数にWebSocketの入力、出力を受け取り、中でやっていることと言えば、ユーザ名と入力、出力を渡し`ChatRoom`の`join`というメソッドを呼び出しているだけである。早速`ChatRoom.java`の`joinメソッド`を見てみよう。

#ChatRoom.java
```java:ChatRoom.java
package models;

import play.mvc.*;
import play.libs.*;
import play.libs.F.*;

import scala.concurrent.Await;
import scala.concurrent.duration.Duration;
import akka.actor.*;
import static akka.pattern.Patterns.ask;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.fasterxml.jackson.databind.node.ArrayNode;



import java.util.*;

import static java.util.concurrent.TimeUnit.*;

/**
 * A chat room is an Actor.
 */
public class ChatRoom extends UntypedActor {
    
    // Default room.
    static ActorRef defaultRoom = Akka.system().actorOf(Props.create(ChatRoom.class));
    
    // Create a Robot, just for fun.
    static {
        new Robot(defaultRoom);
    }
    
    /**
     * Join the default room.
     */
    public static void join(final String username, WebSocket.In<JsonNode> in, WebSocket.Out<JsonNode> out) throws Exception{
        
        // Send the Join message to the room
        String result = (String)Await.result(ask(defaultRoom,new Join(username, out), 1000), Duration.create(1, SECONDS));
        
        if("OK".equals(result)) {
            
            // For each event received on the socket,
            in.onMessage(new Callback<JsonNode>() {
               public void invoke(JsonNode event) {
                   
                   // Send a Talk message to the room.
                   defaultRoom.tell(new Talk(username, event.get("text").asText()), null);
                   
               } 
            });
            
            // When the socket is closed.
            in.onClose(new Callback0() {
               public void invoke() {
                   
                   // Send a Quit message to the room.
                   defaultRoom.tell(new Quit(username), null);
                   
               }
            });
            
        } else {
            
            // Cannot connect, create a Json error.
            ObjectNode error = Json.newObject();
            error.put("error", result);
            
            // Send the error to the socket.
            out.write(error);
            
        }
        
    }
    
    // Members of this room.
    Map<String, WebSocket.Out<JsonNode>> members = new HashMap<String, WebSocket.Out<JsonNode>>();
    
    public void onReceive(Object message) throws Exception {
        
        if(message instanceof Join) {
            
            // Received a Join message
            Join join = (Join)message;
            
            // Check if this username is free.
            if(members.containsKey(join.username)) {
                getSender().tell("This username is already used", getSelf());
            } else {
                members.put(join.username, join.channel);
                notifyAll("join", join.username, "has entered the room");
                getSender().tell("OK", getSelf());
            }
            
        } else if(message instanceof Talk)  {
            
            // Received a Talk message
            Talk talk = (Talk)message;
            
            notifyAll("talk", talk.username, talk.text);
            
        } else if(message instanceof Quit)  {
            
            // Received a Quit message
            Quit quit = (Quit)message;
            
            members.remove(quit.username);
            
            notifyAll("quit", quit.username, "has left the room");
        
        } else {
            unhandled(message);
        }
        
    }
    
    // Send a Json event to all members
    public void notifyAll(String kind, String user, String text) {
        for(WebSocket.Out<JsonNode> channel: members.values()) {
            
            ObjectNode event = Json.newObject();
            event.put("kind", kind);
            event.put("user", user);
            event.put("message", text);
            
            ArrayNode m = event.putArray("members");
            for(String u: members.keySet()) {
                m.add(u);
            }
            
            channel.write(event);
        }
    }
    
    // -- Messages
    
    public static class Join {
        
        final String username;
        final WebSocket.Out<JsonNode> channel;
        
        public Join(String username, WebSocket.Out<JsonNode> channel) {
            this.username = username;
            this.channel = channel;
        }
        
    }
    
    public static class Talk {
        
        final String username;
        final String text;
        
        public Talk(String username, String text) {
            this.username = username;
            this.text = text;
        }
        
    }
    
    public static class Quit {
        
        final String username;
        
        public Quit(String username) {
            this.username = username;
        }
        
    }
    
}
```
`joinメソッド`、最初の一文はこれである。

```java:ChatRoom.java
String result = (String)Await.result(
    ask(defaultRoom,new Join(username, out), 1000), 
    Duration.create(1, SECONDS));
```
`Await.result`、`ask`、`Duration.create`、どれも見慣れないが、それぞれ
`Await.result` は `scala.concurrent.Await.result`、
`ask` は `akka.pattern.Patterns.ask`、
`Duration.create` は `scala.concurrent.duration.Duration.create`
のことである。ちょっとAPIリファレンスを見てみる。

##scala.concurrent.Await.result→[API](http://www.scala-lang.org/files/archive/nightly/docs/library/index.html#scala.concurrent.Await$)

```text
def result[T](awaitable: Awaitable[T], atMost: Duration): T
Await and return the result (of type T) of an Awaitable.
```
処理待ちして、Awaitableの結果(T)を返す、と言っている。
引数にはAwaitable(T)とDurationを受け取ると言っている。

##akka.pattern.Patterns.ask→[API](http://doc.akka.io/api/akka/2.0/akka/pattern/Patterns$.html)
続いてask

```text
 def ask (actor: ActorRef, message: Any, timeoutMillis: Long): Future[AnyRef]

Java API for `akka.pattern.ask`: Sends a message asynchronously and returns a Future holding the eventual reply message; this means that the target actor needs to send the result to the sender reference provided. The Future will be completed with an akka.actor.AskTimeoutException after the given timeout has expired; this is independent from any timeout applied while awaiting a result for this future (i.e. in Await.result(..., timeout)).
```
メッセージを非同期で送り、結果、応答メッセージを含んだ`Future`を受け取ると言っている。
引数には`ActorRef`とメッセージ(`Any`)と`Long`を受け取る、と言っている。

`Future`は`Awaitableインターフェイス`を実装しているようなので、上記`Await.result`の第一引数になれたのだね。

```java:ChatRoom.java
ask(defaultRoom,new Join(username, out), 1000)
```
をもう少し見てみると、
`defaultRoom`→第1引数`ActorRef`
`new Join(username, out)`→第2引数`Any(Javaでいう'Object')`のメッセージ
`1000`→第3引数`Long`のタイムアウト
となっている。

defaultRoomというのはChatRoomクラスの頭の方に定義されているstaticなActorRef型の変数である。

```java:ChatRoom.java
static ActorRef defaultRoom = Akka.system().actorOf(Props.create(ChatRoom.class));
```
`Akka.system` → `play.libs.Akka.system`
`actorOf` → `akka.actor.ActorSystem.actorOf`
`Props.create` → `akka.actor.Props.create `
これらをそれぞれ見ていく。

##play.libs.Akka.system()→[API](http://www.playframework.com/documentation/2.0/api/java/play/libs/Akka.html)
```
public static akka.actor.ActorSystem system()

    Retrieve the application Akka Actor system. 
```
akka.actor.ActorSystemを取ってくると言っている。

##akka.actor.ActorSystem.actorOf→[API](http://doc.akka.io/api/akka/2.0/akka/actor/ActorSystem.html)
```text
 def actorOf (props: Props): ActorRef

Create new actor as child of this context and give it an automatically generated name (currently similar to base64-encoded integer count, reversed and with “$” prepended, may change in the future).
```

このコンテキスト(?)の子として新しいアクターを作成し、自動で名前をつけ、ActorRefを返すと言っている。
Props(以下で説明)を引数として受け取ると言っている。

##akka.actor.Props.create→[API](http://doc.akka.io/japi/akka/2.3-M1/akka/actor/Props.html)
```
 static Props 	create(java.lang.Class<?> clazz, java.lang.Object... args)
          Java API: create a Props given a class and its constructor arguments.
```
Classクラスのオブジェクトを受け取り、Propsを作成すると言っている。

##いったんまとめ
`join`メソッド、最初の一文のこれで

```java:ChatRoom.java
String result = (String)Await.result(
    ask(defaultRoom,new Join(username, out), 1000), 
    Duration.create(1, SECONDS));
```

Actor(ChatRoom)に`Join`型のメッセージを非同期で送信し、処理結果を`Future`型で受け取って(ask)、`Awaitable`型で受け取って(result)、String型に変換している。
端的に言うと、<b>チャットのハブとなっているChatRoomのオブジェクト(Actor)に、入室の旨を伝え、反応を待っている</b>わけである。
`Join`は自分で定義したクラスで、`username`とWebSocketの出力チャンネルが入っている。
ちなみに、`ask`に定義してあるタイムアウトに達しても応答が返らない場合、`akka.actor.AskTimeoutException`が発生する。`Await.result`の第2引数で定義してあるタイムアウトに達しても応答が返らない場合、`java.util.concurrent.TimeoutException`が発生する。微妙に違うので注意。

#続・ChatRoom.java(joinメソッド)
```java:ChatRoom.java
if("OK".equals(result)) {
    
    // For each event received on the socket,
    in.onMessage(new Callback<JsonNode>() {
       public void invoke(JsonNode event) {
           
           // Send a Talk message to the room.
           defaultRoom.tell(new Talk(username, event.get("text").asText()), null);
           
       } 
    });
    
    // When the socket is closed.
    in.onClose(new Callback0() {
       public void invoke() {
           
           // Send a Quit message to the room.
           defaultRoom.tell(new Quit(username), null);
           
       }
    });
    
} else {
    
    // Cannot connect, create a Json error.
    ObjectNode error = Json.newObject();
    error.put("error", result);
    
    // Send the error to the socket.
    out.write(error);
    
}
```
ChatRoomアクターに入室を問い合わせ、すでにユーザ名が使われているなどでエラーになった場合、WebSocketの出力チャンネルにエラー用のJSONを書き出している。これでクライアント側のWebSocketの`onmessage`が反応する。

入室が無事OKだった場合は、WebSocketの入力用チャンネルに`in.onmessage`でクライアントからメッセージを受信した時のハンドラ、`in.onClose`でWebSocketが閉じられた時のハンドラを`play.libs.F`の`Callback`の無名クラスで登録している。

`onMessage`時は、`defaultRoom`の`tell`メソッドで、`Talk`というオブジェクト(チャットのテキストを表す)を、`onClose`時は`Quit`というオブジェクトを渡している。

---
さて、`ask`で送られてきた`Join`も、`tell`で送られてきた`Talk`も`Quit`も、すべて`ChatRoom.java`の`onReceive`で受け止めて処理を行うのだが・・・長くなったのでまた[次回](http://qiita.com/yyyske/items/a6f5090f555038d6ff01)！

---
[Playframework付属のサンプル'websocket-chat'を見てみる(1/4)](http://qiita.com/yyyske/items/d9e3cf69f7a6f5fb732b)
[Playframework付属のサンプル'websocket-chat'を見てみる(2/4)](http://qiita.com/yyyske/items/0d00e5e352b7078a501e)
[Playframework付属のサンプル'websocket-chat'を見てみる(4/4)](http://qiita.com/yyyske/items/a6f5090f555038d6ff01)