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