Angular と Akka Streams で多対多のビデオチャット作った – Media Source Extensions 利用

最近は Akka をよく触っています。Akka Streams でテキストを扱い文字チャットを実装することができたので、今度はバイナリを流してみようと思い立ちました。

拡張機能を利用せずに、HTML5の機能だけで実装できないか調べていたのですが、HTML5 の Media Source Extensions が使えそうだったのでこれを使ってビデオチャットを作ることにしました。この記事では、少しずつ実装を進めていく様子について記載します。実装の細かい部分については、Developers.IO 2017 E-2 トラックで話す予定です。

最終的に作ったもの

バージョン情報

パッケージ名	バージョン
@angular/core	4.1.2
rxjs	5.3.1
typescript	2.2.2
Playframework	2.5.12
Google Chrome（実行環境）	58

進め方の方針

メディアを扱うための道具を選定

ビデオチャットを実装する上で主戦場となるのは絶え間なく流れ続けるバイナリデータをどう扱うかという点です。私はバイナリデータを扱った経験がなく、メディアを取り扱う道具の選定に苦戦しました。リアルタイム Publish / Subscribe については文字チャットを実装したことでだいたいの実装イメージはついていたので、すぐに決まりました。最終的に以下のものを使うことにしました。

採用した道具	役割	採用理由
HTML5 Media Source Extensions (Chrome)	ブラウザ上でメディアを扱うためのAPIとして利用	拡張機能やプラグインレスで実現したかった。また、普段馴染みのある Google Chrome で動かしたかった
Angular	Akka Streams と WebSocket 経由でバイナリデータを授受	Serviceにより通信処理を分離でき、かつオールインワンで楽
Akka Streams	バイナリデータの Publish / Subscribe	WebRTCを回避しつつ多対多のやりとりを実現する手段として

リアルタイムコミュニケーションといえば WebRTC が思いつきますが、STUNサーバやTURNサーバといったコンポーネントを設置するのが大変そうで気が進まなかったため Akka Streams で実装を進めます。

実装の流れ

多対多のビデオチャットを実装するにあたり、いきなり全部やろうとすると確実に進まないだろうということで、いくつかのステップへ分けることにしました。まずは Media Source Extensions を使ってみるところからはじめ、徐々に多対多の仕組みへ近づけていくことにします。

1. 録画した自分の映像をそのまま表示する
2. 録画した自分の映像をサーバー経由で表示する
3. 多対多を実現するために、参加者を動的に管理できるようにする
4. 途中参加へ反応して動的にビデオコンポーネントを作成する

1. 録画した自分の映像をそのまま表示する

Media Source Extensionsが どんなものか使ってみることにします。 MediaDevices.getUserMedia() のページに記載されているサンプルにしたがって、以下の手順を実装しPCカメラで撮影した自分の映像をブラウザへ映します。

• <video>タグを用意する
• MediaDevices.getUserMedia() を実行する。コールバック関数で MediaStream が取得できる
• 取得した MediaStream を<video>タグの srcObject プロパティに設定する

これだけです。 Media Source Extensions の API をつかうことによってずいぶん実装が楽です。

startVideo(): void {

  navigator.mediaDevices.getUserMedia({
    video: true,
    audio: false
  })
    .then((stream) => {
      this.localVideoPlayer.nativeElement.srcObject = stream;
    })
    .catch((error) => {
      console.error('mediaDevice.getUserMedia() error:', error);
      return;
    });

}

<div>
  <button (click)="startVideo();">start</button>
  <video autoplay #localVideoPlayer id="local_video" style="width: 240px; height: 180px; border: 1px solid black; transform: scale(-1, 1);">  </video>
</div>

Webカメラで撮影した映像をそのままブラウザへ表示することができました。

2. 録画した自分の映像をサーバー経由で表示する

次です。 getUserMedia で取得した Stream をそのまま利用する方法はわかりましたが、実際のビデオチャットではサーバを介すことになります。そこで、自分で撮影した映像をサーバ経由で送信・受信し、ブラウザへ表示してみることにします。もしこれができれば、複数人になっても、同じ仕組みで送受信することで 他の人のWeb カメラ映像を映せるはずです。

さて、サーバを経由するとなると先ほどとは同じようにいきません。バイナリデータを取得する必要があります。 Media Source Extensions では、 MediaRecorder を使うことでバイナリを取得することが可能です。

• 投影用の<video id="remote_video">を用意する
• getUserMedia で Stream を取得する
• MediaRecorder オブジェクトを作成する。引数は MediaStream と option オブジェクト
• MediaRecorder の ondataavailable イベントで Blob が取得できるのでこれを利用します
• 取得した Blob を WebSocket にてサーバへ送ります
• WebSocket から受け取ったデータをデコードしてチャンクデータとして <video> タグへ追加していきます

サーバが絡み一気に話が進みます。結果どうなるかまずはごらんください。

サーバに送るところまでは以下のコードで実現できます。

start(): void {

  // -- ①
  let subject = this.videoService.connect(this.roomNumber, this.name);

  // -- ②
  const options = {
    audioBitsPerSecond: 64000,
    videoBitsPerSecond: 100000,
    mimeType: 'video/webm; codecs=vp9'
  };


  this.recorder = new MediaRecorder(this.localStream, options);

  // -- ③
  this.recorder.ondataavailable = (event) => {
    subject.next(event.data);
  };

  this.recorder.onstop = (event) => {
    console.log('recorder.stop(), so playback');
    subject.complete();
    subject = null;
    this.recorder = null;
  };

  // -- ④
  this.recorder.start(500);
  console.log('publish');
}

• ① Rxjs.Subject の生成: WebSocket の関数をバインドした Subjectを作成します
• ② 録画オプション: ビデオのビットレートはChromeが許容する最低にしています。mimeType は Chrome で利用できる WebM を使います
• ③ 録画データの送信: 録画途中、バイナリデータが固まって利用可能になったら、そのままサーバへ送ります
• ④ 録画開始: 引数は録画データを利用可能とする頻度を指定します。500ミリ秒に1回、1秒に2回、すなわち 2fps とします

これでバイナリデータをサーバへ送出するところまでできてました。サーバがやっていることは受け取ったデータをそのまま流すことだけですので、データを受け取る口を用意しそれをデコードします。リアルタイムストリーミングを実現するためにはこの部分も実装でカバーしなければならないことがあるのですが、かなり細かい話になるのでここでは割愛します。 WebM フォーマットでデコードすれば正しく投影できる、ということだけご認識ください。

3. 多対多を実現するために、参加者を動的に管理できるようにする

ここまではすべてソロ活動です。実際にはビデオチャットへの途中参加、途中退出がありえます。各端末には「誰が参加しているか」という情報が必要になります。また、参加しているメンバー全員分のバイナリデータも必要です。よって、受け渡すデータの特性に応じ、別のチャネルを用意します。

• メンバー表（途中入退室に応じて動的に更新）のチャネル
• 各メンバーのバイナリチャネル

チャネルはすべて Akka Streams の MergeHub とBroadcastHub で実装します。（参考：WebSocket のチャットサーバを Akka Streams の MergeHub と BroadcastHub で実装する ｜ Developers.IO）

サーバサイドについては、必要に応じてチャネルを用意する方針で固まりました。ではクライアントサイドはこのチャネルをどのように使えばよいでしょうか。ここで Angular の特性が活きます。メンバー表のリストをHTMLにバインドし、メンバー表の増減に応じてvideoコンポーネントが新しく作られるようにすれば良いです。整理します。

• 自分の名前がついたバイナリチャネルは書き込み専用である。2 で実装した方法を用いて、チャンクデータを WebSocket にて送出すればよい
• メンバー表を Subscribe する。変更に応じて自動でコンポーネントが作られるようにする
• 動的に生成する<video>コンポーネントは、他人の名前がついたチャネルを Subscribe することで映像データのチャンクを取得することができる。このデータを使ってブラウザの <video> コンポーネントにデータをセットする

この仕組で多対多のビデオチャットが実現できるはずです。

export class MembersVideoRoomComponent implements OnInit {

  roomNumber: string;
  name: string;

  private localStream;
  private recorder;
  @ViewChild('localVideoPlayer') localVideoPlayer: any;

  members: Observable<string[]>;

  constructor(
    private videoService: VideoService,
    private memberService: MemberService,
    private route: ActivatedRoute
  ) {
  }

  ngOnInit() {
    this.route.params.forEach((params: Params) => {
      this.roomNumber = params['roomNumber'];
    });

    this.route.queryParams.forEach((params: Params) => {
      this.name = params['name'];
    });

    navigator.mediaDevices.getUserMedia({
      video: true,
      audio: false
    })
      .then((stream) => {
        this.localStream = stream;
        this.localVideoPlayer.nativeElement.srcObject = this.localStream;
      })
      .catch((error) => {
        console.error('mediaDevice.getUserMedia() error:', error);
        return;
      });

    this.members = this.memberService.connect(this.roomNumber, this.name)
      .debounceTime(1000)
      .distinctUntilChanged()
      .concatMap(members => {
        return Observable.of<string[]>(members.members)
      });

  }

<div class="container">
  <button (click)="start();">publish</button>
  <button (click)="stop();">stop</button>

  <ul class="list-group">
    <li class="list-group-item">
      <div class="col-xs-12">
        <video autoplay #localVideoPlayer
               style="width: 240px; height: 180px; border: 1px solid black; transform: scale(-1, 1);"></video>
      </div>
    </li>
  </ul>
  <div class="row container">
    <div *ngFor="let member of members | async" class="col-sm-3">
      <app-member-video [room]=roomNumber [name]=member></app-member-video>
    </div>
  </div>
</div>

これで、途中参加に対応した多対多のチャットを作ることができました…といきたいところでしたが、大きな課題に直面しました。

課題：途中参加したメンバーの画面に、参加済みメンバーのビデオが表示されない

どうやら、 WebM データとその録画方法が影響していそうだということがわかりました。先頭のチャンクと、それ以降のチャンクでは、持っているデータが異なる ということが関係しているようです。

$ ffprobe chunk@3b6d8160 # 先頭データのバイナリを保存したファイル => 正常にメタ情報を表示可能

---
Input #0, matroska,webm, from 'chunk@3b6d8160':
  Metadata:
    encoder         : Chrome
  Duration: N/A, start: 0.000000, bitrate: N/A
    Stream #0:0(eng): Video: vp9 (Profile 0), yuv420p(tv), 640x480, SAR 1:1 DAR 4:3, 1k tbr, 1k tbn, 1k tbc (default)
---

$ ffprobe chunk@4e16e6ea # 先頭より後のバイナリを保存したファイル => メタ情報を表示させようとするとエラー
[mp3 @ 0x7f9e41802a00] Format mp3 detected only with low score of 1, misdetection possible!
[mp3 @ 0x7f9e41802a00] Failed to read frame size: Could not seek to 9100.
chunk@4e16e6ea: Invalid argument

少なくともChromeの場合、途中からのチャンクデータだけでは、<video>の再生を行えない、ということがわかりました。

対策：メンバーの変動があるたびに録画を一度停止し、再開する

途中のデータから再開することができないのならば、必要なタイミングで再度録画しなおすことで、先頭のデータをチャンネルへ再送すればよいだろう、という発想です。結果的にこの作戦はうまくいきました。ただし、チャネルに残っている中途半端なチャンクデータは無視するよう受け取り側でバイナリを解析する必要があるのですが、今回は割愛します。

データ送出側としては、以下のようにすれば良いです。

export class MembersVideoRoomComponent implements OnInit {

  roomNumber: string;
  name: string;

  private localStream;
  private recorder;
  @ViewChild('localVideoPlayer') localVideoPlayer: any;

  members: Observable<string[]>;

  constructor(
    private videoService: VideoService,
    private memberService: MemberService,
    private route: ActivatedRoute
  ) {
  }

  ngOnInit() {
    this.route.params.forEach((params: Params) => {
      this.roomNumber = params['roomNumber'];
    });

    this.route.queryParams.forEach((params: Params) => {
      this.name = params['name'];
    });

    navigator.mediaDevices.getUserMedia({
      video: true,
      audio: false
    })
      .then((stream) => {
        this.localStream = stream;
        this.localVideoPlayer.nativeElement.srcObject = this.localStream;
      })
      .catch((error) => {
        console.error('mediaDevice.getUserMedia() error:', error);
        return;
      });

    this.members = this.memberService.connect(this.roomNumber, this.name)
      .debounceTime(1000)
      .distinctUntilChanged()
      .concatMap(members => {
        this.restart();
        return Observable.of<string[]>(members.members)
      });

  }
  start(): void { ... }
  stop(): void { ... }
  restart(): void {
      if (!this.recorder) {
        return;
      }
      this.stop();
      this.start();
    }
  }

おわりに

Media Source Extensions, Angular, Akka Streams を組み合わせることで、多対多のビデオチャットツールを実装することができました。実用のためにはまだ課題がありますが、実現可能であることがわかったのでよかったです。それと、利用する人の立場になったとき、特別なプラグイン等入れることなくHTML5の機能のみで利用できるという点も嬉しいですね。また、アーキテクチャの観点でみると、 WebRTC が定めるサーバ群を用意せずとも、 Akka Streams ひとつでチャンネルを実現できたということも収穫です。今回のように、

• データチャネルを Akka Streams で実装する
• クライアントとデータチャネルを Rxjs を使ってつなげる

という構成をとれば、リアルタイムの Publish/Subscribe 機構が作れます。応用して他にも作ってみたいと思います。

省略したデコード部分やサーバサイドの実装は Developers.IO 2017 で話します。7月1日（土）開催予定ですのでぜひご参加ください。私担当以外のセッションも面白そうなネタが目白押しです！

• 7/1(土)開催のDevelopers.IO 2017、いよいよ本登録開始！ ｜ Developers.IO

ソースコード

• mediaserver-websocket-akka-stream
• angular-mediastream-app

参考

• Media Source Extensions
• The WebM Project | WebM Container Guidelines
• MediaRecorder と WebM で、オレオレ Live Streaming
• Angular-cli ng serveでhttps - Qiita
• Media Source Extensionsを使ってみた (WebM編) - Qiita
• Web API インターフェイス