add: 添加监听NFD发出的日志信息，并转发到区块链的功能

BlockChainLogger.cpp

@@ -0,0 +1,65 @@
+//
+// Created by sunny on 2020/9/9.
+//
+#include <utility>
+#include "BlockChainLogger.h"
+#include "utils/RingBuffer.h"
+
+
+BlockChainLogger::BlockChainLogger(std::string shareMemoryName,
+                                   std::string host, int port, std::string prefix) :
+    host(std::move(host)), port(port), prefix(std::move(prefix)), face(host, std::to_string(port)),
+    shareMemoryName(std::move(shareMemoryName)) {
+}
+
+void BlockChainLogger::run() {
+  RingBuffer ringBuffer;
+
+  if (!ringBuffer.connect(this->shareMemoryName)) {
+    // 打开共享内存失败，可能是NFD没有启动
+    std::cerr << "连接共享内存: " << shareMemoryName << " => 失败~" << std::endl;
+    exit(-1);
+  }
+  char buffer[4 * 1024 * 1024];
+  int expectedSize = 0;
+  int size = 0;
+  int offset = 0;
+
+  while (true) {
+    if (expectedSize <= 0) {       // 读取一个整型，表示后面跟着的json串的大小
+      size = ringBuffer.read(buffer, 4);
+      if (size <= 0) {    // 没有数据
+//        std::cout << "wait for data" << std::endl;
+        usleep(10000);
+        continue;
+      }
+      expectedSize = *((int *) buffer);
+      offset = 0;
+    }
+    size = ringBuffer.read(buffer + offset, expectedSize);
+    expectedSize -= size;
+    offset += size;
+    if (expectedSize == 0) {   // 读取到完整的一个json串
+      std::string logInfo(buffer, offset);
+      std::cout << logInfo << std::endl;
+
+      // 通过MIN的方式，将日志信息发给区块链
+      Interest interest((Name(prefix)));
+      interest.setMustBeFresh(true);
+      interest.setApplicationParameters((const uint8_t *) buffer, offset);
+      face.expressInterest(interest, [](const Interest &interest1, const Data &data) {
+        // onData
+        std::cout << "onData" << std::endl;
+      }, [](const Interest &interest1, const lp::Nack &nack) {
+        // onNack
+        std::cout << "onNack: " << nack.getReason() << std::endl;
+      }, [](const Interest &interest1) {
+        // onTimeout
+        std::cout << "onTimeout" << std::endl;
+      });
+      face.processEvents();
+    } else {
+//      std::cout << "数据接收不完整" << std::endl;
+    }
+  }
+}

BlockChainLogger.h

@@ -0,0 +1,47 @@
+//
+// Created by sunny on 2020/9/9.
+//
+
+#ifndef MIRCERTIFICATIONMANAGER_BLOCKCHAINLOGGER_H
+#define MIRCERTIFICATIONMANAGER_BLOCKCHAINLOGGER_H
+
+#include <ndn-cxx/face.hpp>
+#include <ndn-cxx/security/key-chain.hpp>
+
+using namespace ndn;
+
+/**
+ * 负责从NFD接收日志信息，并发送给区块链
+ */
+class BlockChainLogger {
+public:
+  /**
+   * 创建区块链日志记录对象
+   * @param host            区块链节点所接入的NFD的ip（如果本地NFD有路由可以转发到区块链所接入的节点，直接连接本地的NFD也可）
+   * @param port            区块链节点所接入的NFD的port（如果本地NFD有路由可以转发到区块链所接入的节点，直接连接本地的NFD也可）
+   * @param prefix          区块链监听在这个前缀接收日志信息
+   */
+  explicit
+  BlockChainLogger(std::string shareMemoryName = "MIN-MIR-WARNING-LOG-TO-BLOCKCHAIN",
+                   std::string host = "localhost", int port = 6363,
+                   std::string prefix = "/mis/log");
+
+  /**
+   * 运行日志转发服务
+   */
+  void run();
+
+private:
+
+
+private:
+  Face face;
+  KeyChain keyChain;
+  std::string prefix;
+  std::string host;
+  std::string shareMemoryName;
+  int port;
+};
+
+
+#endif //MIRCERTIFICATIONMANAGER_BLOCKCHAINLOGGER_H

CMakeLists.txt

@@ -8,9 +8,10 @@ add_executable(MIRCertificationManager main.cpp
         utils/RedisUtil.cpp
         utils/NDNCertificationUtil.cpp
         utils/JsonCppUtil.cpp
+        utils/RingBuffer.cpp
         Server.cpp
-        )
-target_link_libraries(MIRCertificationManager ndn-cxx boost_system hiredis jsoncpp)
+        BlockChainLogger.cpp BlockChainLogger.h)
+target_link_libraries(MIRCertificationManager ndn-cxx boost_system hiredis jsoncpp boost_thread pthread rt)
 
 add_executable(installCertificate installCertificate.cpp
         utils/FileUtils.cpp
@@ -29,3 +30,9 @@ add_executable(testRequest testRequest.cpp
         )
 
 target_link_libraries(testRequest ndn-cxx boost_system)
+
+add_executable(testBlockchainLog testBlockchainLog.cpp
+        utils/RingBuffer.cpp
+        )
+
+target_link_libraries(testBlockchainLog ndn-cxx boost_system hiredis jsoncpp boost_thread pthread rt)

main.cpp

@@ -1,6 +1,8 @@
 #include <iostream>
+#include <thread>
 #include "utils/RedisUtil.h"
 #include "Server.h"
+#include "BlockChainLogger.h"
 
 /**
  *
@@ -18,8 +20,13 @@ using namespace std;
 
 int main() {
 
-    Server server;
-    server.run();
+  Server server;
+  server.run();
+  std::thread t([]() {
+    BlockChainLogger blockChainLogger;
+    blockChainLogger.run();
+  });
+
 
 //    AddOrUpdateCertificateRequestMessage addOrUpdateCertificateRequestMessage;
 //    addOrUpdateCertificateRequestMessage.parse("{\n"

testBlockchainLog.cpp

@@ -0,0 +1,44 @@
+//
+// Created by sunny on 2020/9/9.
+//
+
+#include <iostream>
+#include "utils/RingBuffer.h"
+#include "utils/json.hpp"
+
+int main() {
+  RingBuffer ringBuffer;
+  ringBuffer.open("MIN-MIR-WARNING-LOG-TO-BLOCKCHAIN");
+  nlohmann::json logJsonInfo = {
+      {"Type", "network"},
+      {"Command", "Log"},
+      {"Prefix", "xxx"},
+      {"Level", 1},
+      {"Action", R"({"Action":"","Weight":90,"User":"dmgo--","GateIn":"/min/gdcni16","VRouter":"/min/gdcni11"})"},
+      {"Sig", "xxx"},
+      {"Timestamp", "xxx"}
+  };
+  std::string logInfo = logJsonInfo.dump();
+  int size = logInfo.size();
+  ringBuffer.write((void *) (&size), 4);
+  ringBuffer.write((void *) logInfo.c_str(), logInfo.size());
+  sleep(5);
+  ringBuffer.write((void *) (&size), 4);
+  ringBuffer.write((void *) logInfo.c_str(), logInfo.size() - 2);
+  sleep(2);
+  ringBuffer.write((void *) (logInfo.c_str() + logInfo.size() - 2), 2);
+  sleep(5);
+  ringBuffer.write((void *) (&size), 4);
+  ringBuffer.write((void *) logInfo.c_str(), logInfo.size());
+  sleep(5);
+  ringBuffer.write((void *) (&size), 4);
+  ringBuffer.write((void *) logInfo.c_str(), logInfo.size());
+  sleep(5);
+  ringBuffer.write((void *) (&size), 4);
+  ringBuffer.write((void *) logInfo.c_str(), logInfo.size());
+  sleep(5);
+  ringBuffer.write((void *) (&size), 4);
+  ringBuffer.write((void *) logInfo.c_str(), logInfo.size());
+  sleep(5);
+  ringBuffer.remove();
+}

utils/RingBuffer.cpp

@@ -0,0 +1,195 @@
+//
+// Created by sunny on 2020/6/10.
+//
+
+#include "RingBuffer.h"
+#include <boost/interprocess/mapped_region.hpp>
+#include <boost/interprocess/sync/interprocess_semaphore.hpp>
+#include <semaphore.h>
+#include <cstring>
+
+RingBuffer::RingBuffer(uint64_t bufferSize) :
+        bufferSize(bufferSize) {
+}
+
+RingBuffer::~RingBuffer() {
+    // 移除共享内存
+    sharedMemoryObject.remove(name.c_str());
+}
+
+bool RingBuffer::open(const std::string &shareMemoryName) {
+    try {
+        this->name = shareMemoryName;
+
+        // 创建共享内存
+        sharedMemoryObject = shared_memory_object(create_only, this->name.c_str(), read_write);
+
+        uint64_t shareSize = sizeof(sem_t) + sizeof(RBMeta) + bufferSize;
+
+        // 设置共享内存的大小
+        sharedMemoryObject.truncate(shareSize);
+
+        // 将共享内存映射到当前进程
+        mmap = mapped_region(sharedMemoryObject, read_write);
+
+        this->meta = reinterpret_cast<RBMeta *>((unsigned char *) mmap.get_address() + sizeof(sem_t));
+        this->data = (unsigned char *) mmap.get_address() + sizeof(sem_t) + sizeof(RBMeta);
+
+        // 使用信号量进行同步，保证本次创建所有初始化完成之前，其它想要连接到这个共享内存的进程持续等待，直到初始化完成，发出信号量
+        auto *sem = (sem_t *) mmap.get_address();
+        if (sem_init(sem, 1, 0) == -1) {
+            std::cerr << "初始化信号量失败 => sem_init()" << std::endl;
+            sharedMemoryObject.remove(this->name.c_str());
+            return false;
+        }
+        meta->readIndex = 0;
+        meta->writeIndex = 0;
+        meta->localReadIndex = 0;
+        meta->nextWriteIndex = 0;
+        meta->localWriteIndex = 0;
+        meta->nextReadIndex = 0;
+        memset(this->data, 0, bufferSize);
+        sem_post(sem);
+
+    } catch (...) {
+        std::cerr << "创建共享内存失败" << std::endl;
+        sharedMemoryObject.remove(this->name.c_str());
+        return false;
+    }
+    return true;
+}
+
+bool RingBuffer::connect(const std::string &shareMemoryName) {
+    try {
+        this->name = shareMemoryName;
+
+        // 创建共享内存
+        sharedMemoryObject = shared_memory_object(open_only, this->name.c_str(), read_write);
+
+        uint64_t shareSize = sizeof(sem_t) + sizeof(RBMeta) + bufferSize;
+
+        // 设置共享内存的大小
+        sharedMemoryObject.truncate(shareSize);
+
+        // 将共享内存映射到当前进程
+        mmap = mapped_region(sharedMemoryObject, read_write);
+
+        this->meta = reinterpret_cast<RBMeta *>((unsigned char *) mmap.get_address() + sizeof(sem_t));
+        this->data = (unsigned char *) mmap.get_address() + sizeof(sem_t) + sizeof(RBMeta);
+
+        auto *sem = (sem_t *) mmap.get_address();
+        if (sem_wait(sem) == -1) {
+            std::cerr << "等待信号量失败 => sem_wait()" << std::endl;
+            sharedMemoryObject.remove(this->name.c_str());
+            return false;
+        }
+
+        sem_post(sem);
+    } catch (const std::exception& e) {
+        std::cerr << "连接共享内存失败: " << e.what() << std::endl;
+        sharedMemoryObject.remove(this->name.c_str());
+        return false;
+    }
+    return true;
+}
+
+unsigned char *RingBuffer::operator[](int i) {
+    return this->data + i;
+}
+
+uint64_t RingBuffer::size() {
+//    std::cout << "nextWriteIndex: " << this->meta->nextWriteIndex << ", nextReadIndex：" << this->meta->nextReadIndex
+//              << std::endl;
+    if (this->meta->nextWriteIndex < this->meta->nextReadIndex) {
+        return this->meta->nextWriteIndex - this->meta->nextReadIndex + bufferSize;
+    } else {
+        return this->meta->nextWriteIndex - this->meta->nextReadIndex;
+    }
+}
+
+int RingBuffer::write(void *dataBuffer, uint64_t size) {
+    if (this->meta->isClose) {
+        return -2;
+    }
+    try {
+        auto afterNextWriterIndex = nextVal(this->meta->nextWriteIndex, size);
+        if (this->isWriteExceed(this->meta->nextWriteIndex, size, afterNextWriterIndex, this->meta->localReadIndex)) {
+            // RingBuffer已满，无法再继续写入
+            if (this->isWriteExceed(this->meta->nextWriteIndex, size, afterNextWriterIndex, this->meta->readIndex)) {
+                return -1;
+            }
+
+            // 同步写者记录的localReadIndex和实际的readIndex
+            this->meta->localReadIndex = this->meta->readIndex;
+        }
+
+        if (afterNextWriterIndex < this->meta->nextWriteIndex) {        // 处理回环绕回情况下的写入
+            std::cout << "write 回环" << std::endl;
+            auto firstSize = bufferSize - this->meta->nextWriteIndex;
+            auto remainSize = size - firstSize;
+            memcpy(this->data + this->meta->nextWriteIndex, dataBuffer, firstSize);
+            memcpy(this->data, (void *) (((char *) dataBuffer) + firstSize), remainSize);
+        } else {
+            memcpy(this->data + this->meta->nextWriteIndex, dataBuffer, size);
+        }
+//        std::cout << "write => [" << this->meta->nextWriteIndex << " -> " << size << "]" << std::endl;
+
+        this->meta->nextWriteIndex = afterNextWriterIndex;
+        this->meta->writeIndex = this->meta->nextWriteIndex;
+    } catch (...) {
+        std::cerr << "fuck write error" << std::endl;
+        return -1;
+    }
+    return 0;
+}
+
+int RingBuffer::read(void *dataBuffer, uint64_t size) {
+    if (this->meta->isClose) {
+        return -1;
+    }
+    try {
+//        std::cout << "nextReadIndex: " << this->meta->nextReadIndex << ", localWriteIndex: "
+//                  << this->meta->localWriteIndex << ", writeIndex: " << this->meta->writeIndex << std::endl;
+        if (this->isReadExceed(this->meta->nextReadIndex, this->meta->localWriteIndex)) {
+            // RingBuffer已空，无法再继续写入
+            if (this->isReadExceed(this->meta->nextReadIndex, this->meta->writeIndex)) {
+                return 0;
+            }
+
+            // 同步读者记录的localWriteIndex和实际的writeIndex
+            this->meta->localWriteIndex = this->meta->writeIndex;
+        }
+
+        // 获取当前可读的size
+        auto currentSize = this->size();
+        auto canReadSize = currentSize > size ? size : currentSize;
+        auto afterNextReadIndex = nextVal(this->meta->nextReadIndex, canReadSize);
+
+        if (afterNextReadIndex < this->meta->nextReadIndex) {      // 处理回环情况
+            std::cout << "read 回环" << std::endl;
+            auto firstSize = bufferSize - this->meta->nextReadIndex;
+            auto remainSize = canReadSize - firstSize;
+            memcpy(dataBuffer, this->data + this->meta->nextReadIndex, firstSize);
+            memcpy((void *) (((char *) dataBuffer) + firstSize), this->data, remainSize);
+        } else {
+            memcpy(dataBuffer, this->data + this->meta->nextReadIndex, canReadSize);
+        }
+
+//        std::cout << "read => [" << this->meta->nextReadIndex << "(" << this->meta->writeIndex << ")" << " -> "
+//                  << canReadSize << "]" << std::endl;
+        this->meta->nextReadIndex = this->nextVal(this->meta->nextReadIndex, canReadSize);
+        this->meta->readIndex = this->meta->nextReadIndex;
+        return canReadSize;
+    } catch (...) {
+        std::cerr << "fuck read error" << std::endl;
+    }
+    return -1;
+}
+
+bool RingBuffer::remove() {
+    return sharedMemoryObject.remove(this->name.c_str());
+}
+
+void RingBuffer::close() {
+    this->meta->isClose = true;
+}

utils/RingBuffer.h

@@ -0,0 +1,186 @@
+//
+// Created by sunny on 2020/6/10.
+//
+
+#ifndef MINSOCKETSTACK_CPP_RINGBUFFER_H
+#define MINSOCKETSTACK_CPP_RINGBUFFER_H
+
+#include <string>
+#include <boost/interprocess/managed_shared_memory.hpp>
+#include <iostream>
+
+
+#define CACHE_LINE_SIZE 64      // 缓存行大小
+
+using namespace boost::interprocess;
+
+class RingBuffer {
+
+public:
+    /**
+     * RingBuffer 的元数据结构
+     */
+    struct RBMeta {
+        char cachePad0[CACHE_LINE_SIZE]{};
+        volatile long readIndex{0};
+        volatile long writeIndex{0};
+        char cachePad1[CACHE_LINE_SIZE - 2 * sizeof(long)]{};
+
+        /////////////////////////// For Reader
+        long localWriteIndex{0};
+        long nextReadIndex{0};
+        char cachePad2[CACHE_LINE_SIZE - 2 * sizeof(long)]{};
+        /////////////////////////// End For Reader
+
+        /////////////////////////// For Writer
+        long localReadIndex{0};
+        long nextWriteIndex{0};
+        char cachePad3[CACHE_LINE_SIZE - 2 * sizeof(long)]{};
+        /////////////////////////// End For Writer
+
+        volatile bool isClose{false};  // 标记是否关闭，一个处于Close状态的共享内存对象读返回-1,写返回-2
+    };
+
+    explicit
+    RingBuffer(uint64_t bufferSize = 4 * 1024 * 1024);
+
+    /**
+     * 打开一个共享内存
+     * 1. 打开模式为create_only，如果对应名称的共享内存已经存在，则会返回false
+     * @param shareMemoryName
+     * @return
+     */
+    bool open(const std::string &shareMemoryName);
+
+    /**
+     * 连接一个共享内存
+     * 1. 打开模式为open_only，如果对应名称的共享内存不存在，则会返回false
+     * @param shareMemoryName
+     * @return
+     */
+    bool connect(const std::string &shareMemoryName);
+
+    /**
+     * 移除一个当前创建的共享内存
+     * @return
+     */
+    bool remove();
+
+    /**
+     * 返回当前RingBuffer内数据的大小
+     * @return
+     */
+    uint64_t size();
+
+
+    /**
+     * 获取循环数组的下标
+     * 1. 通过位运算计算，比模运算性能更佳；
+     * 2. 要求bufferSize的大小位2^n
+     *
+     * @param from
+     * @param size
+     * @return
+     */
+    inline uint64_t nextVal(uint64_t from, uint64_t size) const {
+        return (from + size) & (bufferSize - 1);
+    }
+
+    /**
+     * 判断在一个长度为bufferSize的环形数组内，从下标位from位置向前行进length步到达to，会不会经过target(包括等于target)
+     *
+
+     第一种情形：
+     -------------------------------------------------------------------------------------
+         | |                | |             | |         | |                     | |
+     -------------------------------------------------------------------------------------
+          ^                  ^               ^           ^                       ^
+          |                  |               |           |                       |
+        target             from           target        to                      target
+
+     第二种情形（绕过数组终点，回到起点继续走）：
+     -------------------------------------------------------------------------------------
+       | |         | |                | |                               | |         | |
+     -------------------------------------------------------------------------------------
+        ^           ^                  ^                                 ^           ^
+        |           |                  |                                 |           |
+      target       to               target                              from       target
+
+     *
+     * @param from
+     * @param length
+     * @param target
+     * @return
+     */
+    inline bool isExceed(uint64_t from, uint64_t length, uint64_t to, uint64_t target) const {
+        auto result = length >= bufferSize || length <= 0 ||
+                      (
+                              (to > from && target > from && target <= to)
+                              ||
+                              (to < from && (target > from || target <= to))
+                      );
+//        std::cout << "isExceed: [" << from << ", " << to << "], length: " << length << ", target: " << target << " => "
+//                  << result << std::endl;
+        return result;
+    }
+
+    inline bool
+    isWriteExceed(uint64_t nextWriteIndex, uint64_t size, uint64_t afterNextWriteIndex, uint64_t nextReadIndex) const {
+        return size >= bufferSize || size <= 0 ||
+               (
+                       (afterNextWriteIndex > nextWriteIndex && nextReadIndex > nextWriteIndex &&
+                        nextReadIndex <= afterNextWriteIndex)
+                       ||
+                       (afterNextWriteIndex < nextWriteIndex &&
+                        (nextReadIndex > nextWriteIndex || nextReadIndex <= afterNextWriteIndex))
+               );
+    }
+
+    static inline bool
+    isReadExceed(uint64_t nextReadIndex, uint64_t nextWriteIndex) {
+        return nextReadIndex == nextWriteIndex;
+    }
+
+    /**
+     * 往RingBuffer写入数据
+     * @param dataBuffer    数据起始位置
+     * @param size          数据的长度
+     * @return  0 => 写入成功
+     *         -1 => 写入失败（说明RingBuffer已满，请等待一段时间后再次调用）
+     */
+    int write(void *dataBuffer, uint64_t size);
+
+    /**
+     * 在RingBuffer中读取数据
+     * @param dataBuffer    数据存入的起始位置
+     * @param size          最长的长度
+     * @return  -1 => 读取失败，发生错误
+     *         >=0 => 本次读取到的size
+     */
+    int read(void *dataBuffer, uint64_t size);
+
+    /**
+     * 标记关闭该共享内存，不会实际执行销毁操作
+     * 1. 只会在元数据中，把isClose标记为true；
+     * 2. 然后其他人操作同一共享内存区域的时候，期望其读取元数据中isClose的状态；
+     * 3. 如果isClose==true，期望其停止对共享内存的读写操作
+     */
+    void close();
+
+    unsigned char *operator[](int i);
+
+    ~RingBuffer();
+
+private:
+    std::string name;           // 共享内存的名字
+    shared_memory_object sharedMemoryObject;
+    mapped_region mmap;
+    const uint64_t bufferSize{};             // 缓存的大小，只能是2^n => 这样可以使用位运算而不是模运算来计算索引
+
+    ////////////////////下列数据存储在共享内存当中/////////////////////
+    RBMeta *meta{};               // RingBuffer的元数据
+    unsigned char *data{};        // RingBuffer的数据区
+};
+
+
+#endif //MINSOCKETSTACK_CPP_RINGBUFFER_H