ceph/src/rapidjson/example/parsebyparts/parsebyparts.cpp

   1 // Example of parsing JSON to document by parts.
   2
   3 // Using C++11 threads
   4 // Temporarily disable for clang (older version) due to incompatibility with libstdc++
   5 #if (__cplusplus >= 201103L || (defined(_MSC_VER) && _MSC_VER >= 1700)) && !defined(__clang__)
   6
   7 #include "rapidjson/document.h"
   8 #include "rapidjson/error/en.h"
   9 #include "rapidjson/writer.h"
  10 #include "rapidjson/ostreamwrapper.h"
  11 #include <condition_variable>
  12 #include <iostream>
  13 #include <mutex>
  14 #include <thread>
  15
  16 using namespace rapidjson;
  17
  18 template<unsigned parseFlags = kParseDefaultFlags>
  19 class AsyncDocumentParser {
  20 public:
  21     AsyncDocumentParser(Document& d)
  22         : stream_(*this)
  23         , d_(d)
  24         , parseThread_(&AsyncDocumentParser::Parse, this)
  25         , mutex_()
  26         , notEmpty_()
  27         , finish_()
  28         , completed_()
  29     {}
  30
  31     ~AsyncDocumentParser() {
  32         if (!parseThread_.joinable())
  33             return;
  34
  35         {
  36             std::unique_lock<std::mutex> lock(mutex_);
  37
  38             // Wait until the buffer is read up (or parsing is completed)
  39             while (!stream_.Empty() && !completed_)
  40                 finish_.wait(lock);
  41
  42             // Automatically append '\0' as the terminator in the stream.
  43             static const char terminator[] = "";
  44             stream_.src_ = terminator;
  45             stream_.end_ = terminator + 1;
  46             notEmpty_.notify_one(); // unblock the AsyncStringStream
  47         }
  48
  49         parseThread_.join();
  50     }
  51
  52     void ParsePart(const char* buffer, size_t length) {
  53         std::unique_lock<std::mutex> lock(mutex_);
  54
  55         // Wait until the buffer is read up (or parsing is completed)
  56         while (!stream_.Empty() && !completed_)
  57             finish_.wait(lock);
  58
  59         // Stop further parsing if the parsing process is completed.
  60         if (completed_)
  61             return;
  62
  63         // Set the buffer to stream and unblock the AsyncStringStream
  64         stream_.src_ = buffer;
  65         stream_.end_ = buffer + length;
  66         notEmpty_.notify_one();
  67     }
  68
  69 private:
  70     void Parse() {
  71         d_.ParseStream<parseFlags>(stream_);
  72
  73         // The stream may not be fully read, notify finish anyway to unblock ParsePart()
  74         std::unique_lock<std::mutex> lock(mutex_);
  75         completed_ = true;      // Parsing process is completed
  76         finish_.notify_one();   // Unblock ParsePart() or destructor if they are waiting.
  77     }
  78
  79     struct AsyncStringStream {
  80         typedef char Ch;
  81
  82         AsyncStringStream(AsyncDocumentParser& parser) : parser_(parser), src_(), end_(), count_() {}
  83
  84         char Peek() const {
  85             std::unique_lock<std::mutex> lock(parser_.mutex_);
  86
  87             // If nothing in stream, block to wait.
  88             while (Empty())
  89                 parser_.notEmpty_.wait(lock);
  90
  91             return *src_;
  92         }
  93
  94         char Take() {
  95             std::unique_lock<std::mutex> lock(parser_.mutex_);
  96
  97             // If nothing in stream, block to wait.
  98             while (Empty())
  99                 parser_.notEmpty_.wait(lock);
 100
 101             count_++;
 102             char c = *src_++;
 103
 104             // If all stream is read up, notify that the stream is finish.
 105             if (Empty())
 106                 parser_.finish_.notify_one();
 107
 108             return c;
 109         }
 110
 111         size_t Tell() const { return count_; }
 112
 113         // Not implemented
 114         char* PutBegin() { return 0; }
 115         void Put(char) {}
 116         void Flush() {}
 117         size_t PutEnd(char*) { return 0; }
 118
 119         bool Empty() const { return src_ == end_; }
 120
 121         AsyncDocumentParser& parser_;
 122         const char* src_;     //!< Current read position.
 123         const char* end_;     //!< End of buffer
 124         size_t count_;        //!< Number of characters taken so far.
 125     };
 126
 127     AsyncStringStream stream_;
 128     Document& d_;
 129     std::thread parseThread_;
 130     std::mutex mutex_;
 131     std::condition_variable notEmpty_;
 132     std::condition_variable finish_;
 133     bool completed_;
 134 };
 135
 136 int main() {
 137     Document d;
 138
 139     {
 140         AsyncDocumentParser<> parser(d);
 141
 142         const char json1[] = " { \"hello\" : \"world\", \"t\" : tr";
 143         //const char json1[] = " { \"hello\" : \"world\", \"t\" : trX"; // Fot test parsing error
 144         const char json2[] = "ue, \"f\" : false, \"n\": null, \"i\":123, \"pi\": 3.14";
 145         const char json3[] = "16, \"a\":[1, 2, 3, 4] } ";
 146
 147         parser.ParsePart(json1, sizeof(json1) - 1);
 148         parser.ParsePart(json2, sizeof(json2) - 1);
 149         parser.ParsePart(json3, sizeof(json3) - 1);
 150     }
 151
 152     if (d.HasParseError()) {
 153         std::cout << "Error at offset " << d.GetErrorOffset() << ": " << GetParseError_En(d.GetParseError()) << std::endl;
 154         return EXIT_FAILURE;
 155     }
 156
 157     // Stringify the JSON to cout
 158     OStreamWrapper os(std::cout);
 159     Writer<OStreamWrapper> writer(os);
 160     d.Accept(writer);
 161     std::cout << std::endl;
 162
 163     return EXIT_SUCCESS;
 164 }
 165
 166 #else // Not supporting C++11
 167
 168 #include <iostream>
 169 int main() {
 170     std::cout << "This example requires C++11 compiler" << std::endl;
 171 }
 172
 173 #endif