php - ajax 调用从 php libevent 客户端获取连续/流数据

Question

我正在寻找哪个ajax调用或jquery api从php libevent客户端接收流数据。

Libevent 客户端将从 Libevent Server.c 接收数据

服务器.c

/* For socket functions */
#include <sys/socket.h>
#include <event2/event.h>
#include <event2/buffer.h>
#include <event2/bufferevent.h>
#include <unistd.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/un.h>
#include <event2/listener.h>

#include <arpa/inet.h>

#include <signal.h>

#define MAX_LINE 16384

void readcb(struct bufferevent *bev, void *ctx)
{
  /* This callback is invoked when there is data to read on bev. */
  struct evbuffer *input = bufferevent_get_input(bev);
  int len = evbuffer_get_length(input);
  char *data;
  data = malloc(len);
  evbuffer_copyout(input, data, len);
  free(data);
}

void writecb(struct bufferevent *bev, void *ctx)
{
  //here i will be checking my database and memcache new updates
  //it will wait randomly between 2 to 6 seconds to check again

  char *message = "Continuous message from your Server";
  evbuffer_add(bufferevent_get_output(bev), message, strlen(message));
  sleep(2+rand()%4);
}

void errorcb(struct bufferevent *bev, short error, void *ctx)
{
  if (error & BEV_EVENT_EOF) {
    /* connection has been closed, do any clean up here */
    /* ... */
  } else if (error & BEV_EVENT_ERROR) {
    /* check errno to see what error occurred */
    /* ... */
  } else if (error & BEV_EVENT_TIMEOUT) {
    /* must be a timeout event handle, handle it */
    /* ... */
  }
  bufferevent_free(bev);
}

void do_accept(evutil_socket_t listener, short event, void *arg)
{
  struct event_base *base = arg;
  struct sockaddr_storage ss;
  socklen_t slen = sizeof(ss);
  int fd = accept(listener, (struct sockaddr*)&ss, &slen);
  if (fd < 0) {
    perror("accept");
  } else if (fd > FD_SETSIZE) {
    close(fd);
  } else {
    struct bufferevent *bev;
    evutil_make_socket_nonblocking(fd);
    bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE);
    bufferevent_setcb(bev, readcb, writecb, errorcb, NULL);
    bufferevent_setwatermark(bev, EV_READ, 0, MAX_LINE);
    bufferevent_enable(bev, EV_READ|EV_WRITE);
  }
}

void run(void)
{
  evutil_socket_t listener;
  struct sockaddr_un sun;
  int len;
  struct event_base *base;
  struct event *listener_event;
  base = event_base_new();

  if (!base)
    return; /*XXXerr*/

  listener = socket(AF_UNIX, SOCK_STREAM, 0);
  evutil_make_socket_nonblocking(listener);

#ifndef WIN32
{
    int one = 1;
    setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
}
#endif

  memset(&sun, 0, sizeof(sun));

  sun.sun_family = AF_UNIX;
  strcpy(sun.sun_path, "/tmp/influenzaunix.sock");

  unlink(sun.sun_path);
  len = strlen(sun.sun_path) + sizeof(sun.sun_family);
  if (bind(listener, (struct sockaddr *)&sun, len) < 0) {
    perror("bind");
    return;
  }

  if (listen(listener, 16)<0) {
    perror("listen");
    return;
  }

  listener_event = event_new(base, listener, EV_READ|EV_PERSIST, do_accept, (void*)base);

  event_add(listener_event, NULL);

  event_base_dispatch(base);
}

int main(int c, char **v)
{
  setvbuf(stdout, NULL, _IONBF, 0);

  run();
  return 0;
}

编译

gcc -o Server Server.c -levent

那么Client.php是

<?php
define('MAX_LINE', '16384');
define('CHUNK_SIZE', '128');

class Server extends Thread{
    public function run(){
        passthru('./Server');
    }
}

class Client extends Thread{

    public function readcb($bev, $ctx)
    {
        $tmp = array();

        while (1) {
            $data = event_buffer_read($bev,CHUNK_SIZE);
            $size = strlen($data);
            if($size <= 0)
                break;
            $tmp[] = $data;
        }
        $data = implode($tmp);
        echo $data."\n"; //display the data received from server
        flush(); /*flush the data so that an ajax call will be receiving it*/
    }

    public function eventcb($bev, $events, $ptr)
    {
    }

    public function run()
    {
        $sockpath = '/tmp/influenzaunix.sock';
        $socket = stream_socket_client("unix://{$sockpath}",$errorNumber,$errorString,STREAM_CLIENT_CONNECT);
        stream_set_blocking($socket, 0); //non-blocking mode

        if (($socket) === FALSE){
            echo 'connect error';
        }
        else{
            $message = "Send data about this topic"; //subscribing to get data about a topic
            $bev = event_buffer_new($socket, array($this,"readcb"), NULL, array($this,"eventcb"));

            $base = event_base_new();

            if($base !== FALSE)
            {
                event_buffer_base_set($bev,$base);
                event_buffer_enable($bev, EV_READ|EV_WRITE);
                event_buffer_write($bev, $message, strlen($message));
            }

            event_base_loop($base);
        }
    }
}

$server = new Server();
$server->start();

sleep(3); /*let's wait for 3 seconds to make sure Server is ready to accept connection*/    

$client = new Client();
$client->start();
?>

Client.php只能通过CLI模式运行，如下所示的RunClient.php

<?php
passthru('php Client.php');
?>

我面临着提出一个 ajax 调用或 jquery ajax 调用来接收来自 Client.php 的刷新数据的挑战。 JQuery ajax 或 Ajax 调用不能直接调用 Client.php，而是会调用 RunClient.php。如果有人可以尝试提供 ajax 调用或 Jquery ajax 调用来接收刷新的数据，我将不胜感激。

Answer 1

使用浏览器进行测试是了解正在发生的情况的第一步。

我不熟悉 php Thread 类，但是当你调用时可能会发生

passthru('php Client.php');

在 RunClient.php 中，调用处于阻塞状态，等待 php Client.php 完成。

由于Client.php是一个长时间运行的服务器(它基本上永远不会结束)，即使你在Client中使用flush()，你也永远不会得到答案.php.

你可能想尝试

<?php
$fp = fopen('php Client.php');
fpassthru($fp);
?>

查看是否在浏览器中看到结果(确保您的服务器发出数据)

之后，你提到你想使用ajax。您需要了解 ajax 调用有点像浏览器调用:它是请求->响应机制，而不是流机制。

您可以尝试在 html 页面中使用脚本标记

<script src='RunClient.php'/>

并修改您的 Client.php，而不是

echo $data."\n";

与

echo "document.write('<span>".$data."</span>');\n";

以及不带引号的简单数据。如果它有效，您将需要转义数据，以便发出的字符串是有效的 javascript。

当然，这是一个原型(prototype)。这完全取决于您想做什么。

您可能还想使用 websockets 进行流式传输。该协议(protocol)是为流媒体而设计的。比照https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API