Qnx IPC通信PPS

Qnx自带PPS服务，PPS全称Persistent Publish/Subscribe Service，就是常见的P/S通信模式。
Qnx PPS的通信模式是异步的，Publisher和Subscriber也无需关心对方是否存在。

利用Qnx提供的PPS服务，Publisher可以通知多个Subscriber进行某种动作。
该图是QNX官网的PPS示例场景。

• Applications和组件（比如Media、Radio、Phone等等）通过PPS交互消息。
  
  官网链接：
  http://www.qnx.com/developers/docs/7.1/index.html#com.qnx.doc.pps.developer/topic/about.html

使用QNX PPS

Qnx 的PPS，是利用文件的方式实现的。所以使用起来，跟文件的读写差不多。

• Publisher端打开文件，写入文件。

int fd = open( "/pps/example/button", O_RDWR | O_CREAT, S_IRWXU | S_IRWXG | S_IRWXO );

ssize_t len, bytes_written;
len = snprintf(buf, 256, "state::%s\npub1::%d", state ? "on" : "off", count);
bytes_written = write( fd, buf, len );

• Subscriber打开文件，监听文件描述符产生的写入事件，监听可以使用epoll、select、ionotify(qnx提供)等等。
• 打开文件时“？”后面表示打开模式。delta表示增量模式，所谓增量模式，每次只读取PPS数据中变更的部分。

int fd = open( "/pps/example/button?delta", O_RDONLY );
struct pollfd readfds[1];
 

 while (1) {
    readfds[0].fd = fd;
    readfds[0].events = POLLIN;
    
    int result = poll(readfds, 1, timeout);
    if(result != -1 && result != 0) {
        if( readfds[0].revents & POLLRDNORM ){
               // 也可以选择其他读取方式
               // 利用fd的读文件方式有很多种
               num_bytes = read( fd, buf, sizeof(buf) );
               if (num_bytes > 0) {
                   // 解析数据
               }
        }
    } 
 }

• Publisher打开(不存在的时候创建）PPS服务路径下（默认为/pps，可配置）的文件，往文件里写入数据。Subscriber打开要监听的文件（与Publisher一个文件），监听该文件的写入事件，当写入时读取该文件，即可。

完整的例子（QNX官网提供）

• Publisher

#include <stdlib.h>
#include <stdio.h>
#include <unistd.h>
#include <fcntl.h>
#include <string.h>

int main(int argc, char *argv[])
{
   int fd;
   int state = 0;
   char buf[256];
   struct stat stat_buf;
   int count = 0;
   ssize_t len, bytes_written;

   /* Is PPS running? */
   if (stat( "/pps", &stat_buf) != 0)
   {
      if (errno == ENOENT)
         printf ("The PPS server isn't running.\n");
      else
         perror ("stat (/pps)");
      return EXIT_FAILURE;
   }

   /* Create the "button" object (if it doesn't already exist). */
   fd = open( "/pps/example/button", O_RDWR | O_CREAT, S_IRWXU | S_IRWXG | S_IRWXO );
   if ( fd < 0 )
   {
      perror ("Couldn't open /pps/example/button");
      return EXIT_FAILURE;
   }

   /* Loop forever, toggling the state of the button. */
   while ( 1 )
   {
      usleep (500);
      count++;
      len = snprintf(buf, 256, "state::%s\npub1::%d", state ? "on" : "off", count);
      bytes_written = write( fd, buf, len );
      if (bytes_written == -1)
      {
         perror ("write()");
      }
      else if (bytes_written != len)
      {
         printf ("Bytes written: %d String length: %d\n", bytes_written, len);
      }

      if ( state == 0 )
         state = 1;
      else
         state = 0;
   }

   return EXIT_SUCCESS;
}

• Subscriber

#include <stdlib.h>
#include <stdio.h>
#include <unistd.h>
#include <fcntl.h>
#include <string.h>
#include <sys/poll.h>

int main(int argc, char *argv[])
{
   int fd;
   char buf[256];
   ssize_t num_bytes;
   int ret;
   struct pollfd readfds[1];
   int timeout = 2000;

   fd = open( "/pps/example/button?delta", O_RDONLY );
   if ( fd < 0 )
   {
      perror ("Couldn't open /pps/example/button");
      return EXIT_FAILURE;
   }

   /* Loop, echoing the attributes of the button. */
   while (1)
   {
      readfds[0].fd = fd;
      readfds[0].events = POLLIN;

      switch ( ret = poll(readfds, 1, timeout ) )
      {
         case -1:
            /* An error occurred. */
            break;
         case  0:
            /* poll() timed out. */
            break;
         default:
            if( readfds[0].revents & POLLRDNORM )
            {
               num_bytes = read( fd, buf, sizeof(buf) );
               if (num_bytes > 0)
               {
                  write (STDOUT_FILENO, buf, (size_t) num_bytes);
               }
            }
      }
   }

   return EXIT_SUCCESS;
}