// 实现proto文件中服务接口的远程过程调用（RPC）方法
namespace StreamExpample{
class StreamServiceIml : public StreamService {
public:
    StreamServiceIml() {}
    virtual ~StreamServiceIml() {}

    virtual void StreamFun(::google::protobuf::RpcController* controller,
                        const ::StreamExpample::StreamRequest* request,
                        ::StreamExpample::StreamResponse* response,
                        ::google::protobuf::Closure* done) {
        brpc::ClosureGuard done_guard(done);
        brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);

        brpc::StreamOptions stream_options;
        stream_options.handler = &_StreamReceiver;
        if(brpc::StreamAccept(&_streamid, *cntl, &stream_options) != 0) {
            cntl->SetFailed("Fail to accept stream");
            return;
        }
        response->set_message("Server accepted stream");
    }

private:
    brpc::StreamId _streamid;
    StreamReceiver _StreamReceiver;
};
}

class StreamInputHandler {
public:    
    // 当接收到消息后被调用
    virtual int on_received_messages(StreamId id, 
                                     butil::IOBuf *const messages[], 
                                     size_t size) = 0; 
    // 当Stream上长时间没有数据交互后被调用
    virtual void on_idle_timeout(StreamId id) = 0; 
    // 当Stream被关闭时被调用
    virtual void on_closed(StreamId id) = 0;
};

class StreamReceiver : public brpc::StreamInputHandler {
public:
    virtual int on_received_messages(brpc::StreamId id, 
                                     butil::IOBuf *const messages[], 
                                     size_t size) {
        std::ostringstream os;
        for (size_t i = 0; i < size; ++i) {
            os << "msg[" << i << "]=" << *messages[i];
        }
        LOG(INFO) << "Received from Stream=" << id << ": " << os.str();
        return 0;
    }

    virtual void on_idle_timeout(brpc::StreamId id) {
        LOG(INFO) << "Stream=" << id << " has no data transmission for a while";
    }

    virtual void on_closed(brpc::StreamId id) {
        LOG(INFO) << "Stream=" << id << " is closed";
    }
};

brpc::Server server;

StreamExpample::StreamServiceIml stream_servive_iml;
if(server.AddService(&stream_servive_iml, brpc::SERVER_DOESNT_OWN_SERVICE) != 0) {
    LOG(ERROR) << "Fail to add service";
    return -1;
}

brpc::ServerOptions options;
options.idle_timeout_sec = FLAGS_idle_timeout_s;
if(server.Start(FLAGS_port, &options) != 0) {
    LOG(ERROR) << "Fail to start server";
    return -1;
}

#include <butil/iobuf.h>
#include <gflags/gflags.h>
#include <brpc/log.h>
#include <brpc/server.h>
#include "buffer_demo.pb.h"
#include <brpc/stream.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/wait.h>
#include <unistd.h>
#include <iostream>

#define SHARED_MEMORY_KEY 1234
#define SHARED_MEMORY_SIZE 1024

char* sharedMemory;
int shmid;

DEFINE_int32(idle_timeout_s, -1, "Connection will be closed if there is no "
             "read/write operations during the last `idle_timeout_s'");
DEFINE_int32(port, 8000, "TCP Port of this server");

// 实现proto文件中服务接口的远程过程调用（RPC）方法, 这里是存入到共享内存中
namespace BufferDemo{
class BufferServiceIml : public BufferService {
public:
    BufferServiceIml() {}
    virtual ~BufferServiceIml() {}

    virtual void BufferFun(::google::protobuf::RpcController* controller,
                        const ::BufferDemo::Request* request,
                        ::BufferDemo::Response* response,
                        ::google::protobuf::Closure* done) {
        brpc::ClosureGuard done_guard(done);
        brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);

        std::string attachment = cntl->request_attachment().to_string();
        strncpy(sharedMemory, attachment.c_str(), SHARED_MEMORY_SIZE);  
    }
};
}

int main() {
    // 创建共享内存
    shmid = shmget(SHARED_MEMORY_KEY, SHARED_MEMORY_SIZE, IPC_CREAT | 0666);
    if (shmid == -1) {
        std::cerr << "Failed to create shared memory" << std::endl;
        return -1;
    }

    // 附加共享内存
    sharedMemory = (char*)shmat(shmid, NULL, 0);
    if (sharedMemory == (char*)-1) {
        std::cerr << "Failed to attach shared memory" << std::endl;
        return -1;
    }

    brpc::Server server;
    BufferDemo::BufferServiceIml buffer_service_iml;
    if(server.AddService(&buffer_service_iml, brpc::SERVER_DOESNT_OWN_SERVICE) != 0) {
        LOG(ERROR) << "Fail to add service";
        return -1;
    }

    brpc::ServerOptions options;
    options.idle_timeout_sec = FLAGS_idle_timeout_s;
    if(server.Start(FLAGS_port, &options) != 0) {
        LOG(ERROR) << "Fail to start server";
        return -1;
    }

    // 创建子进程
    pid_t pid = fork();
    if (pid < 0) {
        perror("fork");
        exit(1);
    } else if (pid == 0) {
        // 子进程用于读取共享内存数据并清空
        while (1) {
            // 检查共享内存是否有数据
            if (strlen(sharedMemory) > 0) {
                // 读取共享内存中的数据
                printf("Child Process read data from buffer: %s\n", sharedMemory);
                // 清空共享内存
                memset(sharedMemory, 0, SHARED_MEMORY_SIZE);
            }
        }
    }

    server.RunUntilAskedToQuit();
    // 等待子进程结束
    wait(NULL);
    // 分离共享内存
    shmdt(sharedMemory);
    // 删除共享内存
    shmctl(shmid, IPC_RMID, NULL);
    return 0;
}