Go言語で習作:チャットアプリ

今回の目的

前回はTCP/IPの実装法と動作確認を行いました.
今回は, 前回のコードを拡張して複数ユーザが参加可能な簡単なチャットアプリを作ってみたいと思います

なお, 今回作成したコードはこちらのリポジトリにコミットされています

構想

チャットなので, 一人の発言者からのメッセージを参加者全員にリアルタイムで配布したいのですが,
メッセージの受信と送信を一つのプロセスで行う形のままでは, 受信待ちのたびに処理がブロックされてしまい
リアルタイムにメッセージを送信することができません.

そこで, 受信と送信を並列化し, メッセージの送信を受信状況に関わらず行えるようにします.
イメージ図は以下の感じです.

いわゆるオブザーバパターンというやつですね.

では, 早速実装していきましょう!

実装

Client

コード

client

package main

import (
    "fmt"
    "net"
    "os"
    "bufio"
)


func main() {
    connection, error := net.Dial("tcp", "localhost:10000");

    if error != nil {
        panic(error);
    }

    defer connection.Close()
    go waitMessage(connection);

    sendMessage(connection);
}

func sendMessage(connection net.Conn) {
    fmt.Print("> ");

    stdin := bufio.NewScanner(os.Stdin)
    if stdin.Scan() == false {
        fmt.Println("Ciao ciao!");
        return;
    }

    _, error := connection.Write([]byte(stdin.Text()));

    if error != nil {
        panic(error);
    }

    sendMessage(connection)
}

func waitMessage(connection net.Conn) {

    var response = make([]byte, 4 * 1024);
    _, error := connection.Read(response);

    if (error != nil) {
        panic(error);
    }

    fmt.Printf("Server> %s \n", response);

    waitMessage(connection)
}

解説

クライアントのコードは, メッセージの受信と送信を並列化している以外
前回と変わりません

    go waitMessage(connection);

    sendMessage(connection);

処理を並列に行いたい場合はgo構文を使って処理をgoroutine化するだけでOK.
簡単ですね.

Server

コード

/chat/notification.go

package chat

import (
    "net"
)

type NotificationType int

const (
    Message NotificationType = iota
    Join
    Defect
)

// Observerへの通知を意味する構造体
type Notification struct {
    Type NotificationType
    ClientId int
    Connection net.Conn
    Message string
}

/chat/receiver.go

package chat

import (
    "net"
)

type Receiver struct {
    Id int
    Connection net.Conn
    Observer chan<- Notification
}

func (receiver Receiver) Start() {
    receiver.Observer <- Notification{ Type: Join, ClientId: receiver.Id, Connection:receiver.Connection }

    receiver.WaitMessage();
}

func (receiver Receiver) WaitMessage() {
    var buf = make([]byte, 1024);

    n, error := receiver.Connection.Read(buf);
    if (error != nil) {
        receiver.Observer <- Notification{ Type: Defect, ClientId: receiver.Id}
        return;
    }

    receiver.Observer <- Notification{ Type: Message, ClientId: receiver.Id, Message: string(buf[:n])}

    receiver.WaitMessage();
}

/caht/observer.go

package chat

import (
    "net"
    "fmt"
)


type Observer struct {
    Senders []Sender
    Subject <-chan Notification
}

func (observer Observer) WaitNotice() {
    notice := <-observer.Subject

    switch notice.Type {
    case Message:
        for i := range observer.Senders {
            observer.Senders[i].SendMessage(notice.Message);
        }

        break;

    case Join:
        observer.Senders = appendSender(notice.ClientId, notice.Connection, observer.Senders);

        fmt.Printf("Client %d join, now menber count is %d\n", notice.ClientId, len(observer.Senders));
        break;

    case Defect:
        observer.Senders = removeSender(notice.ClientId, observer.Senders);

        fmt.Printf("Client %d defect, now menber count is %d\n", notice.ClientId, len(observer.Senders));
        break;

    default:

    }

    observer.WaitNotice();
}

func appendSender(senderId int, connection net.Conn, senders []Sender) []Sender {
    return append(senders, Sender{ Id: senderId, Connection: connection})
}

func removeSender(senderId int, senders []Sender) []Sender {
    var find = -1;

    for i := range senders {
        if (senders[i].Id == senderId) {
            find = i;
            break;
        }
    }

    if (find == -1) {
        return senders;
    }

    return append(senders[:find], senders[find+1:]...);
}

/chat/sender

package chat

import (
    "net"
)

type Sender struct {
    Id int
    Connection net.Conn
}

func (sender Sender) SendMessage(message string) {
    var buf = []byte(message);

    _, error := sender.Connection.Write(buf)
    if error != nil {
        panic(error);
    }
}

main.go

package main

import (
    "fmt"
    "net"
    "./chat"
)

func main() {

    listener, error := net.Listen("tcp", "localhost:10000");

    if error != nil {
        panic(error);
    }

    fmt.Println("Server running at localhost:10000");

    var channel = make(chan chat.Notification);
    var observer chat.Observer = chat.Observer{ Senders: make([]chat.Sender, 0, 5), Subject: channel};
    go observer.WaitNotice();
    waitClient(listener, 0, observer, channel);

}

func waitClient(listener net.Listener, sequence int, observer chat.Observer, channel chan chat.Notification) {
    connection, error := listener.Accept();

    if error != nil {
        panic(error);
    }

    var receiver chat.Receiver = chat.Receiver{ Id: sequence, Connection: connection, Observer: channel};
    go receiver.Start();

    waitClient(listener, sequence + 1, observer, channel);
}

解説

メッセージの流れの上流から解説ということで, まずはReceiverから.
処理はStartメソッドを呼び出されると同時に, Observerに新たなメンバー入室の通知を行います

func (receiver Receiver) Start() {
    receiver.Observer <- Notification{ Type: Join, ClientId: receiver.Id, Connection:receiver.Connection }

    receiver.WaitMessage();
}

並列化した処理間でのメッセージのやり取りは, メッセージの送信側受信側で共有する
Channel型の変数を使って行います.
ここで使っている, receiver.ObserverもChannelで, あとで解説するmain関数の中で生成されています.

Receiverはクライアントからのメッセージを受信するたびに, Observerにメッセージ受信の通知を行い, 再び受信待ち状態にもどります

    receiver.Observer <- Notification{ Type: Message, ClientId: receiver.Id, Message: string(buf[:n])}

    receiver.WaitMessage();

また, 通信切断を検知したら, メンバー退室の通知を行い, 処理を終了します.

    n, error := receiver.Connection.Read(buf);
    if (error != nil) {
        receiver.Observer <- Notification{ Type: Defect, ClientId: receiver.Id}
        return;
    }

次はObserver.
Observerは, waitNotice関数の中で通知を待ち,

func (observer Observer) WaitNotice() {
    notice := <-observer.Subject

受け取った通知の種別によってメッセージの送信, あるいはメンバーの追加/削除を行います

    switch notice.Type {
    case Message:
        for i := range observer.Senders {
            observer.Senders[i].SendMessage(notice.Message);
        }

        break;

    case Join:
        observer.Senders = appendSender(notice.ClientId, notice.Connection, observer.Senders);

        fmt.Printf("Client %d join, now menber count is %d\n", notice.ClientId, len(observer.Senders));
        break;

    case Defect:
        observer.Senders = removeSender(notice.ClientId, observer.Senders);

        fmt.Printf("Client %d defect, now menber count is %d\n", notice.ClientId, len(observer.Senders));
        break;

    default:

    }

最後に, waitNotice関数を呼び出し, 再び通知待ち状態に戻ります.

    observer.WaitNotice();
}

基本的にObserverはサーバプログラムが生きてる限り走り続ける想定なので, ループから抜ける処理はありません.

そしてSender.

func (sender Sender) SendMessage(message string) {
    var buf = []byte(message);

    _, error := sender.Connection.Write(buf)
    if error != nil {
        panic(error);
    }
}

受け取った文字列をコネクション経由で送信するだけなので, 特に言うことはありません.
今回の設計ではメッセージの送信は並列化していないので, 引数はchannel経由ではなく
直接引数として渡しています.

最後にmain関数の説明をします.
main関数では, net#Listen関数でチャット参加待ち用のポートを確保したのち, Reciver-Observer間のChannelを生成し
Observer#WaitNoticeをgoroutineとして走らせます

    listener, error := net.Listen("tcp", "localhost:10000");

    if error != nil {
        panic(error);
    }

    fmt.Println("Server running at localhost:10000");

    var channel = make(chan chat.Notification);
    var observer chat.Observer = chat.Observer{ Senders: make([]chat.Sender, 0, 5), Subject: channel};
    go observer.WaitNotice();

その後, 自身はチャットへの参加者待ちループに入ります

    waitClient(listener, 0, observer, channel);

}

func waitClient(listener net.Listener, sequence int, observer chat.Observer, channel chan chat.Notification) {
    connection, error := listener.Accept();

    if error != nil {
        panic(error);
    }

    var receiver chat.Receiver = chat.Receiver{ Id: sequence, Connection: connection, Observer: channel};
    go receiver.Start();

    waitClient(listener, sequence + 1, observer, channel);
}

残件

基本的な動きはできるようになりましたが

• エラーハンドリング
• サバクラ間でメッセージ送受信以外の通信を行うためのプロトコル設計
• 複数チャットルームへの対応

などといった拡張まだまだ考えられます.
この辺についてはおいおい考えていこうかと思います.