JavaとgRPCで2Dバーチャルオフィスっぽい何かを作る ③音声をストリーミング

この記事は TLB Enjoy Developers Advent Calendar 2022 の24日目の記事です。

概要

前回に引き続き、2Dバーチャルオフィスっぽいツールを作っていきます。
JavaとgRPCで2Dバーチャルオフィスっぽい何かを作る ②gRPC bidirectional streaming

ちなみに出来上がったものは↓になります。それっぽいものができたのではないでしょうか。

今回の目的

今回はマイクで拾った音声をgRPCでストリーミングして、プレイヤー同士でボイスチャットをできるようにしていきます。
流れとしては以下の通りです。

1. クライアント側のマイクで音声を取得
2. 音声をgRPCでサーバー側に送信する
3. サーバー側は接続したクライアントを保持しておき、音声を受信したら他の接続クライアントに送信する。
4. サーバー側から送信された音声を他のクライアントで受信し、音声を再生する

シーケンスのイメージ図は以下になります。前回の位置同期の時と同様、Bidirectional streamingを使用します。

注意

今回はとりあえず動かしてみた程度の内容になります。
なんちゃって実装のため、ツールを起動したら拾った音声を問答無用で他の全員に配信するようになっています。
これが無用にサーバーに負荷をかけてしまい複数台(3台以上)が接続すると大幅な遅延が発生して使い物になりません。
調べた限り、以下のような対策が考えられるのですが今回はスケジュール的に無理そうだったので見送ってます。ご留意ください。

• そもそも音声や映像をストリーミングするのにgRPCのようなTCP接続は向いていない。UDP接続を使うべき。
• ボイスチャット用のgRPCサーバーを別立てで用意する
• ユーザーが話していないタイミングでは音声を拾うのを停止するような制御を入れる。
• トークルームのような仕組みを作り、配信する相手を制限する。

etc...

実装

ソースコードは以下になります。
https://github.com/kdr250/grpc-2d-sample

protoファイル

Talk.proto

syntax = "proto3";

option java_multiple_files = true;

option java_package = "com.example.shared";

option java_outer_classname = "TalkProto";

service Talk {
  rpc Stream (stream TalkRequest) returns (stream TalkResponse);
}

message TalkRequest {
  int32 readBytes = 1;
  bytes talkByteArray = 2;
}

message TalkResponse {
  int32 readBytes = 1;
  bytes otherTalkByteArray = 2;
}

Serverの実装

接続があったResponseObserverを保持しておき、クライアントから音声のリクエストがあると他のクライアントに配信するようにしています。

TalkService.java

import com.example.shared.TalkGrpc;
import com.example.shared.TalkRequest;
import com.example.shared.TalkResponse;
import io.grpc.stub.StreamObserver;
import org.lognet.springboot.grpc.GRpcService;

import java.util.HashSet;
import java.util.Set;

@GRpcService
public class TalkService extends TalkGrpc.TalkImplBase {

  Set<StreamObserver<TalkResponse>> responseObserverSet = new HashSet<>();

  @Override
  public StreamObserver<TalkRequest> stream(StreamObserver<TalkResponse> responseObserver) {
    responseObserverSet.add(responseObserver);

    return new StreamObserver<TalkRequest>() {
      @Override
      public void onNext(TalkRequest value) {
        responseObserverSet.stream().filter(o -> o != responseObserver).forEach(o -> {
          TalkResponse talkResponse = TalkResponse.newBuilder()
            .setOtherTalkByteArray(value.getTalkByteArray())
            .setReadBytes(value.getReadBytes())
            .build();
          o.onNext(talkResponse);
        });
      }

      @Override
      public void onError(Throwable t) {
        t.printStackTrace();
      }

      @Override
      public void onCompleted() {
        responseObserverSet.remove(responseObserver);
      }
    };
  }
}

Clientの実装

StreamObserverをそのまま使うには要件が複雑すぎる気がしたので、カスタムのStreamObserverを作りました。

TalkResponseObserver.java

import com.example.shared.TalkRequest;
import com.example.shared.TalkResponse;
import com.google.protobuf.ByteString;
import io.grpc.stub.StreamObserver;

import javax.sound.sampled.AudioFormat;
import javax.sound.sampled.AudioSystem;
import javax.sound.sampled.DataLine;
import javax.sound.sampled.SourceDataLine;
import javax.sound.sampled.TargetDataLine;
import java.io.ByteArrayOutputStream;

public class TalkResponseObserver implements StreamObserver<TalkResponse> {

  private StreamObserver<TalkRequest> talkRequestObserver;

  private TargetDataLine microphone;

  private SourceDataLine speakers;

  private final AudioFormat format = new AudioFormat(8000.0f, 16, 1, true, true);

  private static final int CHUNK_SIZE = 1024;

  public TalkResponseObserver() {
    try {
      microphone = AudioSystem.getTargetDataLine(format);
      DataLine.Info info = new DataLine.Info(TargetDataLine.class, format);
      microphone = (TargetDataLine) AudioSystem.getLine(info);
      microphone.open(format);
      microphone.start();

      DataLine.Info dataLineInfo = new DataLine.Info(SourceDataLine.class, format);
      speakers = (SourceDataLine) AudioSystem.getLine(dataLineInfo);
      speakers.open(format);
      speakers.start();
    } catch (Exception e) {
      throw new RuntimeException(e.getMessage(), e);
    }
  }

  @Override
  public void onNext(TalkResponse value) {
    speakers.write(value.getOtherTalkByteArray().toByteArray(), 0, value.getReadBytes());
  }

  @Override
  public void onError(Throwable t) {
    t.printStackTrace();
  }

  @Override
  public void onCompleted() {
    speakers.drain();
    speakers.close();
    microphone.close();
  }

  public void setTalkRequestObserver(final StreamObserver<TalkRequest> talkRequestObserver) {
    this.talkRequestObserver = talkRequestObserver;
  }

  public void sendTalk() {
    int numBytesRead;
    byte[] data = new byte[microphone.getBufferSize() / 5];

    ByteArrayOutputStream out = new ByteArrayOutputStream();
    int bytesRead = 0;

    while (bytesRead < 4000) {
      numBytesRead = microphone.read(data, 0, CHUNK_SIZE);
      bytesRead += numBytesRead;

      out.write(data, 0, numBytesRead);
    }

    ByteString byteString = ByteString.copyFrom(out.toByteArray());
    TalkRequest talkRequest = TalkRequest.newBuilder().setTalkByteArray(byteString).setReadBytes(bytesRead).build();
    talkRequestObserver.onNext(talkRequest);
  }
}

上記のカスタムのStreamObserverを使ってサーバーと相互通信するようにしました。

TalkService.java

import com.example.shared.TalkGrpc.TalkStub;
import com.example.shared.TalkRequest;
import io.grpc.stub.StreamObserver;
import net.devh.boot.grpc.client.inject.GrpcClient;
import org.springframework.stereotype.Service;

@Service
public class TalkService implements Runnable {

  @GrpcClient("server")
  private TalkStub talkStub;

  private TalkResponseObserver talkResponseObserver;

  private Thread talkThread;

  public void startThread() {
    talkResponseObserver = new TalkResponseObserver();
    StreamObserver<TalkRequest> talkRequestObserver = talkStub.stream(talkResponseObserver);
    talkResponseObserver.setTalkRequestObserver(talkRequestObserver);

    talkThread = new Thread(this);
    talkThread.start();
  }

  @Override
  public void run() {
    while (talkThread != null) {
      talkResponseObserver.sendTalk();
    }
  }
}

動作確認

↓は受信した音声を再生している様子です。スピーカーやイヤフォンでもちゃんと聞こえているのでヨシ！

参考にしたサイト

• Java Sound API – Capturing Microphone
• How to Play Sound With Java
• Can gRPC be used for audio-video streaming over Internet?
• gRPC File Upload With Client Streaming
• How to download an image (png file) using Grpc + Java
• Java Record Mic to Byte Array and Play sound
• Java Sound, Getting Started, Part 2, Capture Using Specified Mixer

終わりに

Qiita Advent Calendarの約１ヶ月間で2Dバーチャルオフィスっぽいツールを作ってみたのですが、付け焼き刃では如何ともしがたい技術の壁を感じました。本家の2Dバーチャルオフィスツールは、例えば位置の同期一つとっても非常になめらかに動かしていて改めてすごいなと思いました。

とはいえホビープロジェクト(もしくはクソアプリ)としては上々の成果を出せたのではないかと思います！
「TLB の Developers が Enjoy しながら Advent Calendar を作っていく」という今回のAdvent Calendarの趣旨にピッタリではないでしょうか...!?

というわけでみなさまメリークリスマス！！
明日25日は@ttf1998seiyaさんの記事です!