LG

Le format de base est simple. Language Guide Proto 2 | Protocol Buffers , Language Guide Proto 3 | Protocol Buffers

Ici, on crée deux messages AddSubscriber et RemoveSubscriber pour le pattern publisher. Dans chaque message, chaque champ doit avoir un identifiant numéraire.

syntax = "proto3";

package llgc.protobuf.pattern.publisher;

message AddSubscriber
{
  uint32 id_message = 1;
}

message RemoveSubscriber
{
  uint32 id_message = 1;
}

Il est possible de créer des structures plus complexes avec des sous niveaux et d'importer les messages d'autres fichiers.

syntax = "proto3";

import "pattern_publisher.proto";

package llgc.protobuf.test;

message Tcp
{
  message Msg
  {
    message Test
    {
    }

    oneof data
    {
      llgc.protobuf.pattern.publisher.AddSubscriber add_subscriber = 1;
      llgc.protobuf.pattern.publisher.RemoveSubscriber remove_subscriber = 2;
      Test test = 3;
    }
  }

  repeated Msg msg = 1;
}

Si le fichier .proto est dans un sous dossier, cela dépend le fichier est dans un -I.

Généré dans le dossier tests/pattern/abstract_factory :

/usr/bin/protoc --cpp_out=tests -I tests tests/pattern/abstract_factory/abstract_factory.proto

Généré dans tests :

/usr/bin/protoc --cpp_out=tests -I tests/pattern/abstract_factory tests/pattern/abstract_factory/abstract_factory.proto

• proto2 :
  • Il faut toujours mettre optional pour les champs non répétés. Il est déconseillé d'utiliser un champ required car un champ peut évoluer et devenir optional au profit d'un autre. Ces deux champs ont disparu dans proto3 et optional est appliqué.
  • Mettre [packed=true] pour toutes les répétitions. Présent par défaut avec proto3.
• proto3 :
  • Si un champ devient déprécié mais est encore supporté, ajouter [deprecated=true].
  • Si un champ n'est définitivement plus utilisé, remplacer la ligne avec son identifiant par reserved id;.
  • Les identifiants entre 1 et 15 occupent un octet. Ils sont à privilégier pour les champs répétées. Ceux entre 16 et 2047 sont sur deux octets.

Le protocole se déclare dans un fichier .proto.

Soit une classe Greeter avec une procédure RPC Talk. Avec la présences des deux stream, la communication entre le client et le serveur est bidirectionnelle. Les données échangées sont celles du message de type Rpc.

service Greeter {
  rpc Talk (stream Rpc) returns (stream Rpc) {}
}

Soit l'exemple complet :

syntax = "proto3";

import "pattern_publisher.proto";

package llgc.protobuf.test;

service Greeter {
  rpc Talk (stream Rpc) returns (stream Rpc) {}
}

message Rpc
{
  message Msg
  { 
    message Test
    {
    }
    
    oneof data
    { 
      optional llgc.protobuf.pattern.publisher.AddSubscriber add_subscriber = 1;
      optional llgc.protobuf.pattern.publisher.RemoveSubscriber remove_subscriber = 2;
      optional Test test = 3;
    }
  }
  
  repeated Msg msg = 1;
}

Il faut implémenter une classe héritant la classe de service.

class GreeterImpl : public llgc::protobuf::test::Greeter::Service
{
  // La méthode par défaut renvoie UNIMPLEMENTED.
  // À chaque fois qu'un client appelle cette méthode,
  // le serveur lance un thread pour exécuter la fonction Talk.
  // Il peut y avoir en simultanée autant de threads que de clients.
  // Tant que cette méthode n'est pas terminée, la connexion avec le
  // client reste active.
  // Tous les threads travaillent sur la même instance de la classe.
  ::grpc::Status Talk(::grpc::ServerContext* context,
  ::grpc::ServerReaderWriter< ::llgc::protobuf::test::Rpc,
  ::llgc::protobuf::test::Rpc>* stream) override
  {
    ::llgc::protobuf::test::Rpc message;
    // Tant que le client n'a pas fermé la connexion, on reste dans
    // l'attente de ces messages.
    while (stream->Read(&message))
    {
      // Ici, on répond au message.
      stream->Write(message);
 
      // La réponse est facultative. Ça dépend de ce que la
      // fonction doit faire.
    }
 
    // Fermeture de la connexion à l'initiative du serveur.
    // Dans notre cas précis, le client a déjà terminé puisque
    // stream->Read s'est interrompu.
    return grpc::Status::OK;
  }
}
 
// Pour instancier le serveur
GreeterImpl service;
// Le builder n'a besoin d'exister que jusqu'à la commande BuildAndStart.
ServerBuilder builder;
builder.AddListeningPort("0.0.0.0:1234", grpc::InsecureServerCredentials());
builder.RegisterService(&service);
// La variable service doit exister tant que l'instance du serveur existe.
std::unique_ptr<Server> server(builder.BuildAndStart());
// Lancement du serveur. Fonction bloquante.
// Peut être lancée dans un thread.
server->Wait();
// Pour arrêter le serveur (depuis un autre thread que celui
// ayant lancé la méthode Wait).
server->Shutdown();

// Création du client.
std::shared_ptr<grpc::Channel> channel =
  grpc::CreateChannel("localhost:1234",
  grpc::InsecureChannelCredentials());
std::unique_ptr<llgc::protobuf::test::Greeter::Stub> stub =
  llgc::protobuf::test::Greeter::NewStub(channel);
grpc::ClientContext context;
std::shared_ptr<grpc::ClientReaderWriter<::llgc::protobuf::test::Rpc,
  ::llgc::protobuf::test::Rpc> > stream = stub->Talk(&context);
 
// Communication avec le serveur.
::llgc::protobuf::test::Rpc message;
// Envoie un message au serveur.
stream->Write(message);
// Ferme la communication avec le serveur dans le sens client vers serveur.
stream->WritesDone();
// Attente de la réponse de façon bloquante.
stream->Read(&message);
// Pour arrêter un Read bloquant, il faut lancer depuis un autre thread
context.TryCancel();
// Ferme la communication.
// L'appel à Finish fait que la fonction stream->Read coté serveur
// va échouer mais ne sera plus bloquante.
if (!stream->Finish().ok())
  std::cout << "Erreur" << std::endl;