Привет, я Андрей, работаю Flutter разработчиком в компании Финам.

Продолжим развивать сервис Umka.

Экзамен

На примере реализации кода для проведения "экзамена" мы познакомимся с возможностью технологии gRPC передавать данные в виде потока от клиентского приложения на сервис.

Сценарий экзамена пусть будет таким:

Ученик запрашивает у сервиса "Экзамен", представляющий из себя список вопросов.
Ученик поочерёдно, на каждый вопрос, вводит в консоль ответ, который тут же отображается на серверной стороне сервиса.
Если ответ верен, то сервис к экзаменационной оценке добавляет 1.
После оценки последнего вопроса, сервис отправляет ученику результат.

Добавляем описание

Дополним описание сервиса одним типом Exam и двумя вызовами getExam, takeExam:

syntax="proto3";

message Student {
  int32 id = 1;
  string name = 2;
}

message Question {
  int32 id = 1;
  string text = 2;
}

message Answer {
  int32 id = 1;
  Student student = 2;
  Question question = 3;
  string text = 4;
}

message Evaluation {
  int32 id = 1;
  int32 answerId = 2;
  int32 mark = 3;
}

message AnsweredQuestion {
  Question question = 1;
  string answer = 2;
}

message Exam {
  int32 id = 1;
  repeated Question questions = 2;
}

service Umka {
  rpc getQuestion(Student) returns(Question) {}

  rpc sendAnswer(Answer) returns(Evaluation) {}

  rpc getTutorial(Student) returns (stream AnsweredQuestion) {}

  rpc getExam(Student) returns (Exam) {}

  rpc takeExam(stream Answer) returns(Evaluation) {}
}

Ключевое слово repeated говорит о том, что поле questions содержит список. Для языка Dart это будет List<Question>.

В описании удаленного вызова rpc takeExam(stream Answer) returns(Evaluation) {} аннотация stream перед передаваемым типом Answer сообщает компилятору protoc, что код нужно сгенерировать таким образом, чтобы сервис от клиентского приложения получал ответы в виде потока данных.

Выполним "регенерацию" базового Dart кода:

protoc -I protos/ protos/umka.proto --dart_out=grpc:lib/generated

Дополняем серверный код сервиса

В класс UmkaService добавим реализацию метода getExam для получения вопросов экзамена от сервиса:

  @override
  Future<Exam> getExam(ServiceCall call, Student request) async {
    final exam = Exam()..id = 1;
    exam.questions.addAll(questionsDb);
    return exam;
  }

Клиентскому приложению отправляются все вопросы из нашей "базы".

Реализация метода takeExam потребует чуть больше кода:

  @override
  Future<Evaluation> takeExam(ServiceCall call, Stream<Answer> asnswers) async {
  
    var score = 0;
    
    await for (var answer in asnswers) {
      final isCorrect = getCorrectAnswerById(answer.question.id) == answer.text;
      
      print('Received an answer from ${answer.student.name}\n'
          'for a question: ${answer.question.text}'
          'answer: ${answer.text} is correct: $isCorrect');
          
      if (isCorrect) {
        score++;
      }
    }

    print('The student: ${call.clientMetadata?['student_name']}'
        ' finished exam with the score: $score');
        
    return Evaluation()
      ..id = 1
      ..mark = score;
  }
}

В await for получаем по одному ответы из потока, созданного клиентским приложением.
В переменную var score = 0; добавляем балл за каждый правильный ответ.
После отправки последнего ответа, "клиент" закроет стрим, а мы (сервис) вернём ему оценку.
В консоль выводим полезную информацию по ходу "экзамена".

Кусочек кода call.clientMetadata?['student_name'] показывает пример, как можно получить дополнительную информацию из метаданных отправленных клиентским приложением. По своей сути, "под капотом", это один из заголовков HTTP/2 запроса.

Дополняем код клиента

В UmkaTerminalClient добавим метод takeExam:

  Future<Evaluation> takeExam(Student student) async {
    final exam = await stub.getExam(student);

    final questions = exam.questions;

    final answersStream = StreamController<Answer>();

    final evaluationFuture = stub.takeExam(answersStream.stream,
        options: CallOptions(metadata: {'student_name': '${student.name}'}));

    for (var question in questions) {
      final answer = Answer()
        ..question = question
        ..student = student;

      print('Enter the answer for the question: ${question.text}');
      
      answer.text = stdin.readLineSync()!;

      answersStream.add(answer);

      await Future.delayed(Duration(milliseconds: 1));
    }
    unawaited(answersStream.close());

    return evaluationFuture;
  }

Получаем вопросы.
Создаем поток для передачи ответов.
Устанавливаем соединение, передав на сервис созданный "стрим" и метаданные: 'student_name'.
На каждый полученный вопрос, по очереди, ученик вводит ответ в терминал.
После формирования ответ добавляется в поток: answersStream.add(answer);.
Отправив последний ответ, немедленно закрываем поток данных: unawaited(answersStream.close());
Возвращаем Future с оценкой.

Функция unawated нужна чтобы сказать компилятору, что мы уверены в своих действиях и он не показывал "предупреждение".

Она пока доступна только из библиотеки pedantic, поэтому добавим ее в зависимости:

Выполним команду dart pub get.

Метод обращения к сервису напишем так:

  Future<void> callService(Student student) async {
    final evaluation = await takeExam(student);
    print('${student.name}, your exam score is: ${evaluation.mark}');
    await channel.shutdown();
  }

Дожидаемся окончания "экзамена", чтобы получить оценку.
Выводим результат в консоль.
Закрываем соединение с сервисом.

Запуск экзамена

В разных терминальных окнах стартуем сервис и клиентское приложение

Сервис: dart lib/service.dart
Клиентское приложение: dart lib/client.dart

Демонстрация прохождения "экзамена":

"Студент" Ваня опечатался при ответе на третий вопрос и получил "четвёрку".

Техническое интервью

Мы подошли к самому интересному. Парой-тройкой десятков строчек кода мы реализуем чат, для проведения технического интервью.

Для этого используем возможность gRPC осуществлять двунаправленную потоковую передачу данных от сервиса к клиентскому приложению и обратно в рамках одного HTTP/2 соединения.

В описание сервиса добавим тип:

message InterviewMessage {
  string name = 1;
  string body = 2;
}

И удалённый вызов:

rpc techInterview(stream InterviewMessage) returns(stream InterviewMessage) {}

Вновь выполним "регенерацию" запустив в папке проекта команду:

protoc -I protos/ protos/umka.proto --dart_out=grpc:lib/generated

Код сервиса

На верхний уровень файла lib/service.dart добавим константу с типчными вопросами "собеседования":

const interviewQuestions = [
  'What was wrong in your previous job place?',
  'Why do you want to work for Us?',
  'Who do you see yourself in 5 years?',
  'We will inform you about the decision. Bye!',
];

В класс UmkaService добавим вспомогательную функцию:

  InterviewMessage _createMessage(String text, {String name = 'Interviewer'}) =>
      InterviewMessage()
        ..name = name
        ..body = text;

А также напишем реализацию метода techInterview:

  @override
  Stream<InterviewMessage> techInterview(
      ServiceCall call, Stream<InterviewMessage> interviewStream) async* {
    var count = 0;

    await for (var message in interviewStream) {
      print('Candidate ${message.name} message: ${message.body}');
      if (count >= interviewQuestions.length) {
        return;
      } else {
        yield _createMessage(interviewQuestions[count++]);
      }
    }
  }

После получения первого сообщения от кандидата, что он готов к интервью, отправляем ему вопросы по одному.
На каждый отправленный вопрос дожидаемся ответа.
Получив ответ на последний вопрос, "выходим" — стрим interviewStream на стороне клиента будет закрыт.

Код клиентского приложения

"На клиенте" код метода techInterview пусть будет такой:

  Future<void> techInterview(String candidateName) async {
    final candidateStream = StreamController<InterviewMessage>();
    final interviewerStream = stub.techInterview(candidateStream.stream);

    candidateStream.add(InterviewMessage()
      ..name = candidateName
      ..body = 'I am ready!');

    await for (var message in interviewerStream) {
      print('\nMessage from the ${message.name}:\n${message.body}\n');

      print('Enter your answer:');

      final answer = stdin.readLineSync();

      candidateStream.add(InterviewMessage()..body = answer!);
    }

    unawaited(candidateStream.close());
  }

Создаём стрим final candidateStream = StreamController<InterviewMessage>();.
Передаём candidateStream удаленному вызову, получая обратно interviewerStream.
Отправляем информацию, что кандидат готов к интервью, чтобы завязать "диалог".
Следим за вопросами поступающими в interviewerStream.
Ответы вводим в терминал.
Читаем введённые ответы и отправляем их на сервис.
После завершения потока вопросов от сервиса, закрываем соединение.

К сервису на этот раз будем обращаться так:

  Future<void> callService(Student student) async {
    await techInterview(student.name);
    await channel.shutdown();
  }

Вот гифка демонстрирующая процесс "интервью". В нижнем окне автоматический интервьюер, в верхнем кандидат отвечает на вопросы.

На этом и завершим третью часть где мы, реализовав для нашего сервиса две полезные функции прохождения экзамена и проведения технического интервью, использовали ещё две полезные возможности, предоставляемые технологией gRPC:

Передача данных в виде потока от клиентского приложения к сервису.
Двунаправленная потоковая передача данных между "клиентом" и "сервером" в рамках одного HTTP/2 соединения.

Таким образом, к этому моменту мы успели познакомиться с основными возможностями gRPC.

После перерыва, примерно через месяц-другой, я планирую продолжить данную серию. В планах рассмотреть пример создания простенького мобильного Flutter приложения для работы с сервисом Umka, "деплой" сервиса на реальный сервер, ... .

До встречи в следующей части!

Четвёртая часть

gRPC + Dart, Сервис + Клиент, напишем? Часть 3