简单的用户自定义协议client/server

示例代码

message.h
message.cc
server.cc
client.cc

关于user_defined_protocol

本示例设计一个简单的通信协议，并在协议上构建server和client。server将client发送的消息转换成大写并返回。

协议的格式

协议消息包含一个4字节的head和一个message body。head是一个网络序的整数，指明body的长度。
请求和响应消息的格式一致。

协议的实现

用户自定义协议，需要提供协议的序列化和反序列化方法，这两个方法都是ProtocolMeessage类的虚函数。
另外，为了使用方便，我们强烈建议用户实现消息的移动构造和移动赋值（用于std::move()）。 在ProtocolMessage.h里，序列化反序列化接口如下：

  1. namespace protocol
  2. {
  4. class ProtocolMessage : public CommMessageOut, public CommMessageIn
  5. {
  6. private:
  7.     virtual int encode(struct iovec vectors[], int max);
  9.     /* You have to implement one of the 'append' functions, and the first one
  10.      * with arguement 'size_t *size' is recommmended. */
  11.     virtual int append(const void *buf, size_t *size)；
  12.     virtual int append(const void *buf, size_t size);
  14.     ...
  15. };
  17. }

序列化函数encode

  • encode函数在消息被发送之前调用，每条消息只调用一次。
  • encode函数里，用户需要将消息序列化到一个vector数组，数组元素个数不超过max。目前max的值为8192。
  • 结构体struct iovec定义在请参考系统调用readv和writev。
  • encode函数正确情况下的返回值在0到max之间，表示消息使用了多少个vector。
    • 如果是UDP协议，请注意总长度不超过64k，并且使用不超过1024个vector（Linux一次writev只能1024个vector）。
      • UDP协议只能用于client，无法实现UDP server。
  • encode返回-1表示错误。返回-1时，需要置errno。如果返回值>max，将得到一个EOVERFLOW错误。错误都在callback里得到。
  • 为了性能考虑vector里的iov_base指针指向的内容不会被复制。所以一般指向消息类的成员。

反序列化函数append

  • append函数在每次收到一个数据块时被调用。因此，每条消息可能会调用多次。
  • buf和size分别是收到的数据块内容和长度。用户需要把数据内容复制走。
    • 如果实现了append(const void *buf, size_t *size)接口，可以通过修改*size来告诉框架本次消费了多少长度。收到的size - 消费的size = 剩余的size，剩余的那部分buf会由下一次append被调起时再次收到。此功能更方便协议解析，当然用户也可以全部烤走自行管理，则无需修改*size。
    • 如果是UDP协议，一次append一定是一个完整的数据包。
  • append函数返回0表示消息还不完整，传输继续。返回1表示消息结束。-1表示错误，需要置errno。
  • 总之append的作用就是用于告诉框架消息是否已经传输结束。不要在append里做复杂的非必要的协议解析。

errno的设置

  • encode或append返回-1或其它负数都会被理解为失败，需要通过errno来传递错误原因。用户会在callback里得到这个错误。
  • 如果是系统调用或libc等库函数失败（比如malloc）,libc肯定会设置好errno，用户无需再设置。
  • 一些消息不合法的错误是比较常见的，比如可以用EBADMSG，EMSGSIZE分别表示消息内容错误，和消息太大。
  • 用户可以选择超过系统定义errno范围的值来表示一些自定义错误。一般大于256的值是可以用的。
  • 请不要使用负数errno。因为框架内部用了负数来代表SSL错误。

在我们的示例里，消息的序列化反序列化都非常的简单。
头文件message.h里，声明了request和response类：

  1. namespace protocol
  2. {
  4. class TutorialMessage : public ProtocolMessage
  5. {
  6. private:
  7.     virtual int encode(struct iovec vectors[], int max);
  8.     virtual int append(const void *buf, size_t size);
  9.     ...
  10. };
  12. using TutorialRequest = TutorialMessage;
  13. using TutorialResponse = TutorialMessage;
  15. }

request和response类，都是同一种类型的消息。直接using就可以。
注意request和response必须可以无参数的被构造，也就是说需要有无参数的构造函数，或完全没有构造函数。
message.cc里包含了encode和append的实现：

  1. namespace protocol
  2. {
  4. int TutorialMessage::encode(struct iovec vectors[], int max/*max==8192*/)
  5. {
  6.     uint32_t n = htonl(this->body_size);
  8.     memcpy(this->head, &n, 4);
  9.     vectors[0].iov_base = this->head;
  10.     vectors[0].iov_len = 4;
  11.     vectors[1].iov_base = this->body;
  12.     vectors[1].iov_len = this->body_size;
  14.     return 2;    /* return the number of vectors used, no more then max. */
  15. }
  17. int TutorialMessage::append(const void *buf, size_t size)
  18. {
  19.     if (this->head_received < 4)
  20.     {
  21.         size_t head_left;
  22.         void *p;
  24.         p = &this->head[this->head_received];
  25.         head_left = 4 - this->head_received;
  26.         if (size < 4 - this->head_received)
  27.         {
  28.             memcpy(p, buf, size);
  29.             this->head_received += size;
  30.             return 0;
  31.         }
  33.         memcpy(p, buf, head_left);
  34.         size -= head_left;
  35.         buf = (const char *)buf + head_left;
  37.         p = this->head;
  38.         this->body_size = ntohl(*(uint32_t *)p);
  39.         if (this->body_size > this->size_limit)
  40.         {
  41.             errno = EMSGSIZE;
  42.             return -1;
  43.         }
  45.         this->body = (char *)malloc(this->body_size);
  46.         if (!this->body)
  47.             return -1;
  49.         this->body_received = 0;
  50.     }
  52.     size_t body_left = this->body_size - this->body_received;
  54.     if (size > body_left)
  55.     {
  56.         errno = EBADMSG;
  57.         return -1;
  58.     }
  60.     memcpy(this->body, buf, body_left);
  61.     if (size < body_left)
  62.         return 0;
  64.     return 1;
  65. }
  67. }

encode的实现非常简单，固定使用了两个vector，分别指向head和body。需要注意iov_base指针必须指向消息类的成员。
append需要保证4字节的head接收完整，再读取message body。而且我们并不能保证第一次append一定包含完整的head，所以过程略为繁琐。
append实现了size_limit功能，超过size_limit的会返回EMSGSIZE错误。用户如果不需要限制消息大小，可以忽略size_limit这个域。
由于我们要求通信协议是一来一回的，所谓的“TCP黏包”问题不需要考虑，直接当错误消息处理。
现在，有了消息的定义和实现，我们就可以建立server和client了。　

server和client的定义

有了request和response类，我们就可以建立基于这个协议的server和client。前面的示例里我们介绍过Http协议相关的类型定义：

  1. using WFHttpTask = WFNetworkTask<protocol::HttpRequest,
  2.                                  protocol::HttpResponse>;
  3. using http_callback_t = std::function<void (WFHttpTask *)>;
  5. using WFHttpServer = WFServer<protocol::HttpRequest,
  6.                               protocol::HttpResponse>;
  7. using http_process_t = std::function<void (WFHttpTask *)>;

同样的，对这个Tutorial协议，数据类型的定义并没有什么区别：

  1. using WFTutorialTask = WFNetworkTask<protocol::TutorialRequest,
  2.                                      protocol::TutorialResponse>;
  3. using tutorial_callback_t = std::function<void (WFTutorialTask *)>;
  5. using WFTutorialServer = WFServer<protocol::TutorialRequest,
  6.                                   protocol::TutorialResponse>;
  7. using tutorial_process_t = std::function<void (WFTutorialTask *)>;

server端

server与普通的http server没有什么区别。我们优先IPv6启动，这不影响IPv4的client请求。另外限制请求最多不超过4KB。
代码请自行参考server.cc

client端

client端的逻辑是从标准IO接收用户输入，构造出请求发往server并得到结果。
为了简单，读取标准输入的过程都在callback里完成，因此我们会先发出一条空请求。同样为了安全我们限制server回复包不超4KB。
client端唯一需要了解的就是怎么产生一个自定义协议的client任务，在WFTaskFactory.h有三个接口可以选择：

  1. template<class REQ, class RESP>
  2. class WFNetworkTaskFactory
  3. {
  4. private:
  5.     using T = WFNetworkTask<REQ, RESP>;
  7. public:
  8.     static T *create_client_task(TransportType type,
  9.                                  const std::string& host,
  10.                                  unsigned short port,
  11.                                  int retry_max,
  12.                                  std::function<void (T *)> callback);
  14.     static T *create_client_task(TransportType type,
  15.                                  const std::string& url,
  16.                                  int retry_max,
  17.                                  std::function<void (T *)> callback);
  19.     static T *create_client_task(TransportType type,
  20.                                  const URI& uri,
  21.                                  int retry_max,
  22.                                  std::function<void (T *)> callback);
  23.     ...
  24. };

其中，TransportType指定传输层协议，目前可选的值包括TT_TCP，TT_UDP，TT_SCTP和TT_TCP_SSL。
三个接口的区别不大，在我们这个示例里暂时不需要URL，我们用域名和端口来创建任务。
实际的调用代码如下。我们派生了WFTaskFactory类，但这个派生并非必须的。

  1. using namespace protocol;
  3. class MyFactory : public WFTaskFactory
  4. {
  5. public:
  6.     static WFTutorialTask *create_tutorial_task(const std::string& host,
  7.                                                 unsigned short port,
  8.                                                 int retry_max,
  9.                                                 tutorial_callback_t callback)
  10.     {
  11.         using NTF = WFNetworkTaskFactory<TutorialRequest, TutorialResponse>;
  12.         WFTutorialTask *task = NTF::create_client_task(TT_TCP, host, port,
  13.                                                        retry_max,
  14.                                                        std::move(callback));
  15.         task->set_keep_alive(30 * 1000);
  16.         return task;
  17.     }
  18. };

可以看到我们用了WFNetworkTaskFactory类来创建client任务。
接下来通过任务的set_keep_alive()接口，让连接在通信完成之后保持30秒，否则，将默认采用短连接。
client的其它代码涉及的知识点在之前的示例里都包含了。请参考client.cc

内置协议的请求是怎么产生的

现在系统中内置了http, redis，mysql，kafka四种协议。我们可以通过相同的方法产生一个http或redis任务吗？比如：

  1. WFHttpTask *task = WFNetworkTaskFactory<protocol::HttpRequest, protocol::HttpResponse>::create_client_task(...);

需要说明的是，这样产生的http任务，会损失很多的功能，比如，无法根据header来识别是否用持久连接，无法识别重定向等。
同样，如果这样产生一个MySQL任务，可能根本就无法运行起来。因为缺乏登录认证过程。
一个kafka请求可能需要和多台broker有复杂的交互过程，这样创建的请求显然也无法完成这一过程。
可见每一种内置协议消息的产生过程都远远比这个示例复杂。同样，如果用户需要实现一个更多功能的通信协议，还有许多代码要写。