Class Handler

speech/api/streaming_transcribe_singlethread.cc:35–152

)""";

class Handler : public std::enable_shared_from_this<Handler> {
 public:
  static std::shared_ptr<Handler> Create(google::cloud::CompletionQueue cq,
                                         ParseResult args) {
    return std::shared_ptr<Handler>(
        new Handler(std::move(cq), std::move(args)));
  }

  g::future<g::Status> Start(speech::SpeechClient& client) {
    // Get ready to write audio content.  Create the stream, and start it.
    stream_ = client.AsyncStreamingRecognize();
    // The stream can fail to start; `.get()` returns `false` in this case.
    if (!stream_->Start().get()) return StartFailure();
    // Write the first request, containing the config only.
    if (!stream_->Write(request_, grpc::WriteOptions{}).get()) {
      // Write().get() returns false if the stream is closed.
      return StartFailure();
    }
    auto self = shared_from_this();
    // This creates a series of reads; when each read completes successfully a
    // new one starts.
    stream_->Read().then([self](auto f) { self->OnRead(f.get()); });
    // This creates a series of timer -> write -> timer -> ... steps.
    cq_.MakeRelativeTimer(std::chrono::seconds(1)).then([self](auto f) {
      self->OnTimer(f.get().status());
    });
    return done_.get_future();
  }

 private:
  Handler(google::cloud::CompletionQueue cq, ParseResult args)
      : cq_(std::move(cq)), file_(args.path, std::ios::binary) {
    auto& streaming_config = *request_.mutable_streaming_config();
    *streaming_config.mutable_config() = std::move(args.config);
  }

  g::future<g::Status> StartFailure() {
    auto self = shared_from_this();
    writing_ = false;
    reading_ = false;
    stream_->Finish().then([self](auto f) { self->OnFinish(f.get()); });
    return done_.get_future();
  }

  void OnFinish(g::Status s) { done_.set_value(s); }

  void OnTimer(g::Status s) {
    // On a timer error the completion queue is not usable so just return.
    if (!s.ok()) return;
    auto self = shared_from_this();
    auto constexpr kChunkSize = 64 * 1024;
    std::vector<char> chunk(kChunkSize);
    file_.read(chunk.data(), chunk.size());
    auto const bytes_read = file_.gcount();
    if (bytes_read > 0) {
      request_.clear_streaming_config();
      request_.set_audio_content(chunk.data(), bytes_read);
      std::cout << "Sending " << bytes_read / 1024 << "k bytes." << std::endl;