20.07.2026 21:04
SergeyPyankov 0 9300 Источник

На текущий момент можно считать, что не имеется некоммерческих библиотек или компонентов, позволяющих создать качественный gRPC-клиент (автор, конечно же, в курсе про «DelphiGrpc», но в нынешнем виде полагает её малопригодной для практического использования – хотя бы из-за устаревших версий зависимостей); во многом остроту данной проблемы снимают платные решения – давно существующий компонент TipwGRPC, а также относительно недавно появившийся TsgcGRPCClient. В связи с этим, если по каким-то причинам применение коммерческих вариантов невозможно, то потребуется реализовать указанный клиент своими силами, для чего в статье рассмотрен лишь начальный (но довольно-таки важный) этап разработки: а как, собственно говоря, использующий клиент станет вызывать gRPC-методы – посредством каких функций или процедур, какие параметры они будут требовать, и далее в таком духе? В грубом приближении, что должно быть в этом прототипе вместо многоточий?

function CallMethod(...): ...;

Надо заметить, что представленное здесь рассчитано больше не на универсальные gRPC-клиенты, а скорее специализированные – когда работа всегда ведётся либо с одним и тем же сервером, либо с несколькими похожими и без кардинальной разницы – их возможные нюансы можно не выносить в настройки клиента, заложив непосредственно в код реализации, а оставшиеся немногочисленные параметры указывать прямо в CallMethod.

Также хочется проговорить явно, чтобы читатель не имел ложных надежд, – материал совершенно не содержит ответа на вопрос «как», тут только предложен вариант программного интерфейса к gRPC-клиенту (весьма подробно, однако, расписанный).

Допущение о реализации

Хотя кому-то это может показаться очевидным, но далее предполагается, что gRPC-клиент функционирует в отдельном потоке, им же и созданном (к примеру, в проекте автора их даже два – один отвечает за TCP-соединение и TLS-шифрование, а второй собственно за gRPC поверх HTTP2-транспорта); речь не идёт о том, что разработчик самостоятельно и в обязательном порядке создаёт клиент в каком-то неглавном потоке, а имеется в виду, что поток порождается самим клиентом и недоступен явно извне, другим. Одно из основных преимуществ данного подхода просто́ – главный Delphi-поток при использовании gRPC-методов не блокируется, ведь сценарий их вызова именно из него наверняка самый предпочтительный для большинства.

Виды gRPC-методов

Перейдём к конкретике, для чего, опираясь на поддержку gRPC 4-х видов методов, определим Delphi-интерфейс:

IgRPC = interface
  function CallUnaryMethod(const ServiceName, Name: string; ...): ...;
  function CallServerStreamingMethod(const ServiceName, Name: string; ...): ...;
  function CallClientStreamingMethod(const ServiceName, Name: string; ...): ...;
  function CallBidirectionalStreamingMethod(const ServiceName, Name: string; ...): ...;
end;

Все эти функции неблокирующие (асинхронные). Также здесь сразу же представлены два очевиднейших параметра, без которых немыслим ни один gRPC-вызов, – названия сервиса и вызываемого у него метода. Оформлены же функции не в классе отчасти для того, чтобы полностью абстрагироваться от их реализации (которой, как говорилось в самом начале, не будет).

Оставшиеся параметры потребуют гораздо больших пояснений, нежели возвращаемые функциями типы (причём один параметр даже станет ссылаться на них), поэтому первым делом рассматриваются именно результаты функций: все они тоже будут интерфейсами, позволяющими контролировать вызванный gRPC-метод, причём, что принципиально, стандартный для Delphi механизм управления интерфейсными ссылками (их жизнью) не должен влиять на выполнение метода, т. е. допустимо не сохранять возвращённый интерфейс – это только лишит возможности управлять методом, но не прервёт его.

Возвращаемые интерфейсы

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

IMethod = interface
  procedure Cancel;
end;

Многократный вызов Cancel разрешён, причём даже на уже завершённом gRPC-методе, но вот ещё один нюанс работы с процедурой вынужденно описан ниже.

Возврат именно интерфейса не догма, – однако на практике часто удобнее работать именно с ним, – поэтому в качестве альтернативы можно, скажем, предоставлять числовой идентификатор инициированного gRPC-метода, на основании которого затем выполнять нужные действия, а именно:

IgRPC = interface
  function CallUnaryMethod(...): Integer;
  ...

  procedure Cancel(const MethodID: Integer);
end;

Унарный метод

gRPC не накладывает на этот вид метода требований поддерживать ещё какие-либо действия, кроме только что упомянутой возможности прерывания (отмены), поэтому возвращаться станет именно только что рассмотренный IMethod:

IgRPC = interface
  function CallUnaryMethod(...): IMethod;
  ...
end;

Серверный потоковый метод

Брат-близнец унарного в этом аспекте:

IgRPC = interface
  ...
  function CallServerStreamingMethod(...): IMethod;
  ...
end;

Прочие потоковые методы

Оставшаяся пара самая интересная, ибо позволяет отправить на сервер произвольное количество сообщений (в терминах Protocol Buffers), представляющих собой набор «сырых» байт, полученных после упаковки (сериализации) типизированной структуры данных, в связи с чем тип, описывающий сообщение, достаточно определить как псевдоним для стандартного System.SysUtils.TBytes:

TMessage = TBytes;

Таким образом, IMethod в наследнике будет расширен ещё одной процедурой, а IgRPC примет такой вид:

IStreamingMethod = interface(IMethod)
  procedure Send(const Message: TMessage);
end;

IgRPC = interface
  ...
  function CallClientStreamingMethod(...): IStreamingMethod;
  function CallBidirectionalStreamingMethod(...): IStreamingMethod;
end;

В отличие от Cancel, попытка использования Send на завершённом gRPC-методе должна приводить к исключению.

Основные параметры

Вернёмся к параметризации при порождении gRPC-методов.

Запрос

Кроме уже озвученных ServiceName и Name, необходимо сориентироваться с тем, который станет содержать сообщение, включающее в себя сведения-параметры для вызываемого gRPC-метода (речь о предметно-специфичных данных, обычно необходимых серверу для выполнения метода Name); в gRPC для него принят термин «запрос», поэтому не станем отходить от традиции:

IgRPC = interface
  function CallUnaryMethod(...; const Request: TMessage): ...;
  function CallServerStreamingMethod(...; const Request: TMessage): ...;
  function CallClientStreamingMethod(...; const Request: TMessage): ...;
  function CallBidirectionalStreamingMethod(...; const Request: TMessage): ...;
end;

Результат

Далее определимся с главнейшим параметром – функцией обратного вызова, отвечающей как за обработку успешного результата (для унарного и клиентского потокового методов) и множественных приходящих от сервера сообщений (в случае оставшихся методов), так и информирующей о сбое. Она представлена не обычным процедурным типом навроде function ... of object, а выбор сделан в пользу анонимной процедуры, отличающейся большей универсальность, гибкостью – в неё, при такой необходимости, очень легко обернуть метод объекта, выступающий в качестве обработчика gRPC-вызова, а вот обратное, как правило, весьма редко требуется.

Всего понадобится два типа анонимных процедур (оба раскрытых чуть позже), в зависимости от вида gRPC-метода:

IgRPC = interface
  function CallUnaryMethod(...; const Callback: TResponseCallback): ...;
  function CallServerStreamingMethod(...; const Callback: TResponseCallback): ...;
  function CallClientStreamingMethod(...; const Callback: TStreamingResponseCallback): ...;
  function CallBidirectionalStreamingMethod(...; const Callback: TStreamingResponseCallback): ...;
end;

Все процедуры исполняются в контексте gRPC-потока и не должны содержать объёмных вычислений, а тем более длительных непрогнозируемых ожиданий какого-то ресурса (также следует иметь в виду, что их вызов может состояться ещё до возврата из функции IgRPC); если же результат gRPC-метода неинтересен, то в параметр Callback допустимо передать nil.

Очень важной особенностью является следующее: если анонимная процедура выполняется на момент вызова Cancel, то он должен дождаться её завершения – другими словами, желание прервать gRPC-метод может привести на какое-то время к блокировке потока, задействовавшего возможность отмены, – такова цена гарантии, что Callback после Cancel никогда более не выполнится.

Объявления анонимных процедур отличаются лишь типом их параметра Method:

TResponseCallback = reference to procedure (const Response: TMessage;
  const Method: IMethod; var Error: Exception);

TStreamingResponseCallback = reference to procedure (const Response: TMessage;
  const Method: IStreamingMethod; var Error: Exception);

Трактование этой тройки параметров таково:

  1. Response – для унарного и клиентского потокового методов содержит результат их выполнения, а для прочих – очередное полученное сообщение. При этом, если gRPC-метод, любого вида, завершён с ошибкой (параметр Error не nil), то значение Response не определено и не должно никак использоваться (при реализации самым разумным, в такой ситуации, видится передача сюда пустого массива).

    Как уже встречалось выше, этот массив байт требует обработки – в данном случае распаковки (десериализации) в какую-то типизированную структуру данных.

  2. Method – тот же интерфейс, что возвращён функцией IgRPC. В унарных и клиентских потоковых методах практического смысла не имеет, т. к. сейчас у IMethod наличествует лишь процедура Cancel, а вышеозначенный Callback вызывается как раз в качестве реакции на окончание gRPC-метода; однако, если в дальнейшем IMethod расширится (на что есть немалая вероятность), то не придётся дописывать данный параметр в уже существующие анонимные процедуры обратного вызова (например, так и просится добавить в IMethod процедуру Retry, повторно инициирующую gRPC-метод).

  3. Error – если не nil, то содержит причину сбоя. Возможность изменить параметр (речь о var-модификаторе) сделана для случаев, когда объект нужно проанализировать вне анонимной процедуры (причём под анализом вполне себе может скрываться просто возбуждение исключения на его основе, как показано в примере ниже). При работе с Error действует правило: если присвоить ему значение nil, то ответственность за уничтожение объекта-исключения ложится на разработчика, если же параметр не меняется, то заботиться о вызове его деструктора нужды нет. Таким образом, вариант выноса исключения за пределы процедуры может выглядеть так (в нём некое поле FSomeMethodError имеет тип Exception):

    gRPCClient.CallUnaryMethod
  (
  'SomeService', 'SomeMethod', SomeRequest,
  procedure (const Response: TMessage; const Method: IMethod; var Error: Exception)
  begin
    if Assigned(Error) then
    begin
      TInterlocked.Exchange(FSomeMethodError, Error);
      Error := nil;
    end
    else
      ...
  end
  );

    После чего FSomeMethodError используется в ином месте:

    var
  ErrorBuffer: Exception;
begin
  ...

  ErrorBuffer := TInterlocked.Exchange<Exception>(FSomeMethodError, nil);

  if Assigned(ErrorBuffer) then
    raise ErrorBuffer;

  ...
end;

    Если же обработка сбоя gRPC-метода делается внутри анонимной процедуры, то, само-собой, допустимо применять как способ с условными конструкциями, например такой

    procedure (const Response: TMessage; const Method: IMethod; var Error: Exception)
begin
  if Assigned(Error) then
  begin
    if (Error is ESomeException) and {дополнительные условия} then
      ...
    else
      if (Error is EAnotherException) and {дополнительные условия} then
        ...
  end
  else
    ...
end

    так и задействовать часто более удобный except-блок

    procedure (const Response: TMessage; const Method: IMethod; var Error: Exception)
begin
  if Assigned(Error) then
    try
      try
        raise Error;
      finally
        Error := nil;
      end;
    except
      on ESomeException do ...;
      ...
    end
  else
    ...
end

Таймаут

Финальный параметр у gRPC-методов сравнительно прост – это серверный таймаут, выраженный в миллисекундах (брать меньший или больший интервал – микро- или просто секунды – обычно непрактично, т. к. взаимодействие через gRPC чаще всего ведётся в Интернет, где типичные публичные серверы отрабатывают вызовы за десятки или сотни миллисекунд); тип для выражения таймаута следующий:

const
  NoTimeout = 0;
type
  TMillisecondsTimeout = NoTimeout..99999999;

Максимальное значение в виде 99 999 999 обусловлено gRPC-спецификацией, допускающей не более 8-и цифр. По умолчанию вызов делается без ограничения по длительности:

IgRPC = interface
  function CallUnaryMethod(...; const Timeout: TMillisecondsTimeout = NoTimeout): ...;
  function CallServerStreamingMethod(...; const Timeout: TMillisecondsTimeout = NoTimeout): ...;
  function CallClientStreamingMethod(...; const Timeout: TMillisecondsTimeout = NoTimeout): ...;
  function CallBidirectionalStreamingMethod(...; const Timeout: TMillisecondsTimeout = NoTimeout): ...;
end;

Необязательные параметры

Рассматриваемые здесь опциональные параметры gRPC-методов безотносительны к их виду и понадобиться могут далеко не всем разработчикам и при обращении не к каждому серверу – всего речь пойдёт о паре таких.

Второй таймаут

Только что был разобран таймаут, обрабатываемый лишь сервером – на клиентской стороне он никак не учитывается, поэтому гипотетически возможна ситуация, когда вторая сторона не уложилась в отведённое время, но уведомление об этом клиентом получено не было (скажем из-за сетевых неполадок или даже по причине дефектов в коде gRPC-сервера) – это, к сожалению, приведёт к бесконечному ожиданию итогов gRPC-метода; данную проблему, если с таковой всё же пришлось столкнуться, призван решить клиентский таймаут – это, очевидно, тот интервал времени, в течение которого должен прийти результат gRPC-вызова (логично, что означенный таймаут всегда обязан быть больше серверного как минимум на удвоенную задержку (пинг) до другой стороны). В функциях IgRPC (на примере одной, т. к. в остальных всё так же) он указывается аналогично:

IgRPC = interface
  function CallUnaryMethod(...; const ClientTimeout: TMillisecondsTimeout = NoTimeout): ...;
  ...
end;

Защита от взаимоблокировки

Реальная работа с gRPC-клиентом редко ограничивается вызовом одного-двух методов, интерфейсы для управления которыми можно просто сохранить в паре переменных или полей, – обычно активно (т. е. ожидает результат от сервера) множество методов, чьи интерфейсы необходимо где-то складировать, поэтому пусть, для примера, они хранятся в переменной такого типа:

var
  Methods: TList<IMethod>;

Рано или поздно, скажем по команде от пользователя, понадобится некоторые из хранящихся здесь методов отменить (прервать) и, соответственно, удалить из списка – задача тривиальнейшая и особенностей при текущих вводных не имеет. Всё усложняется, если обращаться к Methods нужно и из анонимных процедур обратного вызова (хотя бы для того, чтобы удалить интерфейс из списка после завершения метода), что сразу делает данную переменную разделяемым (общим) ресурсом, требующим защиты от одновременного доступа из нескольких потоков – того, который используется gRPC-клиентом, и того, которым управляет разработчик. Если говорить языком кода, то в анонимной процедуре видится нечто такое:

gRPCClient.CallUnaryMethod
  (
  'SomeService', 'SomeMethod', SomeRequest,
  procedure (const Response: TMessage; const Method: IMethod; var Error: Exception)
  begin
    ...

    TMonitor.Enter(Methods);
    try
      ... - работа с Methods
    finally
      TMonitor.Exit(Methods);
    end;
  end
  );

Тогда как в другом потоке выполняется отмена gRPC-вызовов:

TMonitor.Enter(Methods);
try
  Methods.Pack
    (
    function (const L, R: IMethod): Boolean
    begin
      Result := {Условие прерывания L};

      if Result then
        L.Cancel;
    end
    );
finally
  TMonitor.Exit(Methods);
end;

Как говорилось ранее, особенность Cancel в том, что если на момент его вызова выполнялась Callback, то он сначала будет дожидаться её завершения – чего никогда не произойдёт в данном сценарии, потому что анонимная процедура станет ждать освобождения того же списка Methods, уже захваченного.

Один из способов избавления от рассмотренной взаимоблокировки заключается в отказе от самостоятельного (явного) захвата ресурса в анонимной процедуре, для чего предлагается передать этот ресурс параметром, а gRPC-клиент сам захватит его перед вызовом Callback и освободит после:

IgRPC = interface
  function CallUnaryMethod(...; const AntiDeadlock: TObject = nil): ...;
  ...
end;

В результате, Cancel в примере выше никогда не будет ожидать, а код анонимной процедуры сократится:

gRPCClient.CallUnaryMethod
  (
  'SomeService', 'SomeMethod', SomeRequest,
  procedure (const Response: TMessage; const Method: IMethod; var Error: Exception)
  begin
    ...

    ... - работа с Methods
  end,
  NoTimeout, NoTimeout, Methods
  );

В случаях же иных, когда требующий защиты ресурс не является классом (допустим представляет собой динамический массив) или вообще не единственный, применяется стандартный для работы с TMonitor приём – создаётся вспомогательный экземпляр TObject, который и передаётся в параметр AntiDeadlock.

Для удобства, все описанные в статье интерфейсы и прочие типы скомпонованы в виде модуля.
unit gRPC;

interface

uses
  System.SysUtils;

const
  NoTimeout = 0;

type
  TMessage = TBytes;
  TMillisecondsTimeout = NoTimeout..99999999;

  IMethod = interface
    procedure Cancel;
  end;

  IStreamingMethod = interface(IMethod)
    procedure Send(const Message: TMessage);
  end;

  TResponseCallback = reference to procedure (const Response: TMessage;
    const Method: IMethod; var Error: Exception);
  TStreamingResponseCallback = reference to procedure (const Response: TMessage;
    const Method: IStreamingMethod; var Error: Exception);

  IgRPC = interface
    function CallUnaryMethod(
      const ServiceName, Name: string;
      const Request: TMessage; const Callback: TResponseCallback;
      const Timeout: TMillisecondsTimeout = NoTimeout;
      const ClientTimeout: TMillisecondsTimeout = NoTimeout;
      const AntiDeadlock: TObject = nil): IMethod;
    function CallServerStreamingMethod(
      const ServiceName, Name: string;
      const Request: TMessage; const Callback: TResponseCallback;
      const Timeout: TMillisecondsTimeout = NoTimeout;
      const ClientTimeout: TMillisecondsTimeout = NoTimeout;
      const AntiDeadlock: TObject = nil): IMethod;
    function CallClientStreamingMethod(
      const ServiceName, Name: string;
      const Request: TMessage; const Callback: TStreamingResponseCallback;
      const Timeout: TMillisecondsTimeout = NoTimeout;
      const ClientTimeout: TMillisecondsTimeout = NoTimeout;
      const AntiDeadlock: TObject = nil): IStreamingMethod;
    function CallBidirectionalStreamingMethod(
      const ServiceName, Name: string;
      const Request: TMessage; const Callback: TStreamingResponseCallback;
      const Timeout: TMillisecondsTimeout = NoTimeout;
      const ClientTimeout: TMillisecondsTimeout = NoTimeout;
      const AntiDeadlock: TObject = nil): IStreamingMethod;
  end;

implementation

end.

Комментарии (0)