Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : * Copyright Contributors to the OpenSDN project
4 : */
5 :
6 : //
7 : // sandesh_session.cc
8 : //
9 : // Sandesh session
10 : //
11 :
12 : #include <boost/bind/bind.hpp>
13 : #include <boost/assign.hpp>
14 : #include <boost/algorithm/string.hpp>
15 :
16 : #include <base/parse_object.h>
17 :
18 : #include <sandesh/common/vns_types.h>
19 : #include <sandesh/common/vns_constants.h>
20 : #include <sandesh/transport/TBufferTransports.h>
21 : #include <sandesh/protocol/TXMLProtocol.h>
22 : #include <sandesh/protocol/TJSONProtocol.h>
23 : #include "sandesh/sandesh_types.h"
24 : #include "sandesh/sandesh.h"
25 :
26 : #include "sandesh_connection.h"
27 : #include "sandesh_session.h"
28 :
29 :
30 : using namespace std;
31 : using namespace contrail::sandesh::protocol;
32 : using namespace contrail::sandesh::transport;
33 : using namespace boost::placeholders;
34 :
35 : using boost::asio::mutable_buffer;
36 : using boost::asio::buffer_cast;
37 :
38 : const std::string SandeshWriter::sandesh_open_ = sXML_SANDESH_OPEN;
39 : const std::string SandeshWriter::sandesh_open_attr_length_ =
40 : sXML_SANDESH_OPEN_ATTR_LENGTH;
41 : const std::string SandeshWriter::sandesh_close_ = sXML_SANDESH_CLOSE;
42 :
43 : //
44 : // SandeshWriter
45 : //
46 142 : SandeshWriter::SandeshWriter(SandeshSession *session)
47 142 : : session_(session),
48 142 : ready_to_send_(true),
49 142 : send_buf_(new uint8_t[kDefaultSendSize]),
50 142 : send_buf_offset_(0) {
51 142 : }
52 :
53 142 : SandeshWriter::~SandeshWriter() {
54 142 : delete [] send_buf_;
55 142 : }
56 :
57 0 : void SandeshWriter::WriteReady(const boost::system::error_code &ec) {
58 0 : if (ec) {
59 0 : SANDESH_LOG(ERROR, "SandeshSession Write error value: " << ec.value()
60 : << " category: " << ec.category().name()
61 : << " message: " << ec.message());
62 0 : session_->increment_write_ready_cb_error();
63 0 : return;
64 : }
65 :
66 : {
67 0 : std::scoped_lock lock(send_mutex_);
68 0 : ready_to_send_ = true;
69 0 : }
70 :
71 : // We may want to start the Runner for the send_queue
72 0 : session_->send_queue()->MayBeStartRunner();
73 : }
74 :
75 12973 : void SandeshWriter::SendMsg(Sandesh *sandesh, bool more) {
76 12973 : SandeshHeader header;
77 12973 : std::stringstream ss;
78 : uint8_t *buffer;
79 12973 : int32_t xfer = 0, ret;
80 : uint32_t offset;
81 : boost::shared_ptr<TMemoryBuffer> btrans(
82 12973 : new TMemoryBuffer(kEncodeBufferSize));
83 : boost::shared_ptr<TXMLProtocol> prot(
84 12973 : new TXMLProtocol(btrans));
85 : // Populate the header
86 12973 : header.set_Namespace(sandesh->scope());
87 12973 : header.set_Timestamp(sandesh->timestamp());
88 12973 : header.set_Module(sandesh->module());
89 12973 : header.set_Source(sandesh->source());
90 12973 : header.set_Context(sandesh->context());
91 12973 : header.set_SequenceNum(sandesh->seqnum());
92 12973 : header.set_VersionSig(sandesh->versionsig());
93 12973 : header.set_Type(sandesh->type());
94 12973 : header.set_Hints(sandesh->hints());
95 12973 : header.set_Level(sandesh->level());
96 12973 : header.set_Category(sandesh->category());
97 12973 : header.set_NodeType(sandesh->node_type());
98 12973 : header.set_InstanceId(sandesh->instance_id());
99 :
100 : // Write the sandesh open envelope.
101 12973 : buffer = btrans->getWritePtr(sandesh_open_.length());
102 12973 : memcpy(buffer, sandesh_open_.c_str(), sandesh_open_.length());
103 12973 : btrans->wroteBytes(sandesh_open_.length());
104 : // Write the sandesh header
105 12973 : if ((ret = header.write(prot)) < 0) {
106 0 : SANDESH_LOG(ERROR, __func__ << ": Sandesh header write FAILED: " <<
107 : sandesh->Name() << " : " << sandesh->source() << ":" <<
108 : sandesh->module() << ":" << sandesh->instance_id() <<
109 : " Sequence Number:" << sandesh->seqnum());
110 0 : session_->increment_send_msg_fail();
111 0 : Sandesh::UpdateTxMsgFailStats(sandesh->Name(), 0,
112 : SandeshTxDropReason::HeaderWriteFailed);
113 0 : sandesh->Release();
114 0 : return;
115 : }
116 12973 : xfer += ret;
117 : // Write the sandesh
118 12973 : if ((ret = sandesh->Write(prot)) < 0) {
119 0 : SANDESH_LOG(ERROR, __func__ << ": Sandesh write FAILED: "<<
120 : sandesh->Name() << " : " << sandesh->source() << ":" <<
121 : sandesh->module() << ":" << sandesh->instance_id() <<
122 : " Sequence Number:" << sandesh->seqnum());
123 0 : session_->increment_send_msg_fail();
124 0 : Sandesh::UpdateTxMsgFailStats(sandesh->Name(), 0,
125 : SandeshTxDropReason::WriteFailed);
126 0 : sandesh->Release();
127 0 : return;
128 : }
129 12973 : xfer += ret;
130 : // Write the sandesh close envelope
131 12973 : buffer = btrans->getWritePtr(sandesh_close_.length());
132 12973 : memcpy(buffer, sandesh_close_.c_str(), sandesh_close_.length());
133 12973 : btrans->wroteBytes(sandesh_close_.length());
134 : // Get the buffer
135 12973 : btrans->getBuffer(&buffer, &offset);
136 : // Sanity
137 12973 : assert(sandesh_open_.length() + xfer + sandesh_close_.length() ==
138 : offset);
139 : // Update the sandesh open envelope length;
140 12973 : char prev = ss.fill('0');
141 : // Adjust for '">'
142 12973 : ss.width(sandesh_open_.length() - sandesh_open_attr_length_.length() - 2);
143 12973 : ss << offset;
144 12973 : ss.fill(prev);
145 12973 : memcpy(buffer + sandesh_open_attr_length_.length(), ss.str().c_str(),
146 25946 : ss.str().length());
147 :
148 : // Update sandesh stats
149 12973 : Sandesh::UpdateTxMsgStats(sandesh->Name(), offset);
150 12973 : session_->increment_send_msg();
151 :
152 12973 : if (send_buf()) {
153 12973 : if (more) {
154 : // There are more messages in the send_queue_.
155 : // Try to package as many sandesh messages as possible
156 : // (== kEncodeBufferSize) before transporting to the
157 : // receiver.
158 6877 : SendMsgMore(btrans);
159 : } else {
160 : // send_queue_ is empty. Flush sandesh->send_buf_ and this message.
161 6096 : SendMsgAll(btrans);
162 : }
163 : } else {
164 : // Send the message
165 0 : SendInternal(btrans);
166 : }
167 12973 : sandesh->Release();
168 12973 : }
169 :
170 : // Package as many sandesh messages as possible [not more than
171 : // kEncodeBufferSize] before transporting it to the receiver.
172 : //
173 : // send_buf_ => unsent data (partial/complete message).
174 : // buf => new message
175 : // buf_len => buf's len
176 6877 : void SandeshWriter::SendMsgMore(boost::shared_ptr<TMemoryBuffer>
177 : send_buffer) {
178 : uint8_t *buf;
179 : uint32_t buf_len;
180 :
181 6877 : send_buffer->getBuffer(&buf, &buf_len);
182 :
183 6877 : if (send_buf_offset()) {
184 : // We have some unsent data.
185 4630 : size_t bulk_msg_len = send_buf_offset() + buf_len;
186 4630 : if (bulk_msg_len < kDefaultSendSize) {
187 : // We still have space for more data. Don't send the message yet.
188 : // Add the message to the existing data.
189 4413 : append_send_buf(buf, buf_len);
190 : } else {
191 : uint8_t *buffer;
192 : // (send_buf_offset() + buf_len) >= kDefaultSendSize
193 : boost::shared_ptr<TMemoryBuffer> bulk_msg(
194 217 : new TMemoryBuffer(send_buf_offset()));
195 217 : buffer = bulk_msg->getWritePtr(send_buf_offset());
196 : // Copy unsent data
197 217 : memcpy(buffer, send_buf(), send_buf_offset());
198 217 : bulk_msg->wroteBytes(send_buf_offset());
199 : // Send it
200 217 : SendInternal(bulk_msg);
201 : // Cleanup send_buf
202 217 : reset_send_buf();
203 : // Next send the new message
204 217 : SendInternal(send_buffer);
205 217 : }
206 : } else {
207 : // We don't have any unsent data.
208 2247 : if (buf_len >= kDefaultSendSize) {
209 : // We don't have room to accommodate anything more.
210 : // Send the message now.
211 17 : SendInternal(send_buffer);
212 : } else {
213 : // We have room to accommodate more data.
214 : // Save this message.
215 : // Note: The memcpy here can be avoided by passing send_buf_
216 : // to TMemoryBuffer so that the message is encoded in
217 : // send_buf_ itself.
218 2230 : set_send_buf(buf, buf_len);
219 : }
220 : }
221 6877 : }
222 :
223 : // sandesh->send_queue_ is empty.
224 : // Flush unsent data (if any) and this message.
225 6096 : void SandeshWriter::SendMsgAll(boost::shared_ptr<TMemoryBuffer> send_buffer) {
226 : uint8_t *buf;
227 : uint32_t buf_len;
228 :
229 6096 : send_buffer->getBuffer(&buf, &buf_len);
230 :
231 6096 : if (send_buf_offset()) {
232 : // We have some unsent data.
233 2013 : size_t bulk_msg_len = send_buf_offset() + buf_len;
234 2013 : if (bulk_msg_len <= kDefaultSendSize) {
235 : uint8_t *buffer;
236 : // We have enough room to send all the pending data in one message.
237 : boost::shared_ptr<TMemoryBuffer> bulk_msg(new
238 1960 : TMemoryBuffer(bulk_msg_len));
239 1960 : buffer = bulk_msg->getWritePtr(bulk_msg_len);
240 1960 : memcpy(buffer, send_buf(), send_buf_offset());
241 1960 : memcpy(buffer + send_buf_offset(), buf, buf_len);
242 1960 : bulk_msg->wroteBytes(bulk_msg_len);
243 : // send the message
244 1960 : SendInternal(bulk_msg);
245 : // reset send_buf_
246 1960 : reset_send_buf();
247 1960 : } else {
248 : uint8_t *buffer;
249 : // We don't have enough space to accommodate all the
250 : // pending data in one message.
251 : boost::shared_ptr<TMemoryBuffer> old_buf(new
252 53 : TMemoryBuffer(send_buf_offset()));
253 53 : buffer = old_buf->getWritePtr(send_buf_offset());
254 53 : memcpy(buffer, send_buf(),
255 : send_buf_offset());
256 53 : old_buf->wroteBytes(send_buf_offset());
257 : // Take care of the unsent data in send_buf_ first.
258 : // Note that we could have accomodated part of buf [last message]
259 : // here. But, not doing so to avoid additional memcpy() :)
260 53 : SendInternal(old_buf);
261 : // Cleanup send_buf_
262 53 : reset_send_buf();
263 : // Well, send the last message now.
264 53 : SendInternal(send_buffer);
265 53 : }
266 : } else {
267 : // No unsent data. Send the message now.
268 4083 : SendInternal(send_buffer);
269 : }
270 6096 : }
271 :
272 6600 : void SandeshWriter::SendInternal(boost::shared_ptr<TMemoryBuffer> buf) {
273 : uint8_t *buffer;
274 : uint32_t len;
275 6600 : buf->getBuffer(&buffer, &len);
276 6600 : std::scoped_lock lock(send_mutex_);
277 6600 : ready_to_send_ = session_->Send((const uint8_t *)buffer, len, NULL);
278 6600 : }
279 :
280 : //
281 : // SandeshSession
282 : //
283 142 : SandeshSession::SandeshSession(SslServer *client, SslSocket *socket,
284 142 : int task_instance, int writer_task_id, int reader_task_id) :
285 : SslSession(client, socket),
286 142 : instance_(task_instance),
287 142 : writer_(new SandeshWriter(this)),
288 142 : reader_(new SandeshReader(this)),
289 284 : send_queue_(new Sandesh::SandeshQueue(writer_task_id,
290 : task_instance,
291 : boost::bind(&SandeshSession::SendMsg, this, _1),
292 142 : kQueueSize)),
293 142 : stats_client_(NULL),
294 142 : connection_(NULL),
295 142 : keepalive_idle_time_(kSessionKeepaliveIdleTime),
296 142 : keepalive_interval_(kSessionKeepaliveInterval),
297 142 : keepalive_probes_(kSessionKeepaliveProbes),
298 142 : tcp_user_timeout_(kSessionTcpUserTimeout),
299 142 : reader_task_id_(reader_task_id),
300 284 : sending_level_(SandeshLevel::INVALID) {
301 142 : if (Sandesh::role() == Sandesh::SandeshRole::Collector) {
302 0 : send_buffer_queue_.reset(new Sandesh::SandeshBufferQueue(writer_task_id,
303 : task_instance,
304 0 : boost::bind(&SandeshSession::SendBuffer, this, _1)));
305 0 : send_buffer_queue_->SetStartRunnerFunc(boost::bind(&SandeshSession::SessionSendReady, this));
306 : }
307 142 : send_queue_->SetStartRunnerFunc(boost::bind(&SandeshSession::SessionSendReady, this));
308 142 : }
309 :
310 284 : SandeshSession::~SandeshSession() {
311 284 : }
312 :
313 12192 : bool SandeshSession::SessionSendReady() {
314 24384 : return (IsEstablished() && writer_->SendReady() &&
315 24384 : Sandesh::IsSendQueueEnabled());
316 : }
317 :
318 840 : void SandeshSession::SetSendQueueWaterMark(
319 : Sandesh::QueueWaterMarkInfo &swmi) {
320 840 : WaterMarkInfo wm(boost::get<0>(swmi),
321 : boost::bind(&SandeshSession::SetSendingLevel, this, _1,
322 1680 : boost::get<1>(swmi)));
323 840 : if (boost::get<2>(swmi)) {
324 420 : send_queue_->SetHighWaterMark(wm);
325 : } else {
326 420 : send_queue_->SetLowWaterMark(wm);
327 : }
328 840 : }
329 :
330 0 : void SandeshSession::ResetSendQueueWaterMark() {
331 0 : send_queue_->ResetHighWaterMark();
332 0 : send_queue_->ResetLowWaterMark();
333 0 : }
334 :
335 142 : void SandeshSession::SetSendingLevel(size_t count, SandeshLevel::type level) {
336 142 : if (sending_level_ != level) {
337 0 : sending_level_ = level;
338 : }
339 142 : }
340 :
341 13449 : SandeshLevel::type SandeshSession::SendingLevel() const {
342 13449 : return sending_level_;
343 : }
344 :
345 142 : void SandeshSession::Shutdown() {
346 142 : if (Sandesh::role() == Sandesh::SandeshRole::Collector) {
347 0 : send_buffer_queue_->Shutdown();
348 : }
349 142 : send_queue_->Shutdown();
350 142 : }
351 :
352 684 : std::string SandeshSession::ToString() const {
353 684 : std::stringstream out;
354 684 : out << TcpSession::ToString() << "(" << instance_ << ")";
355 1368 : return out.str();
356 684 : }
357 :
358 238 : boost::system::error_code SandeshSession::SetSocketOptions() {
359 238 : boost::system::error_code ec = TcpSession::SetSocketOptions();
360 238 : if (ec) {
361 0 : return ec;
362 : }
363 238 : return SetSocketKeepaliveOptions(keepalive_idle_time_, keepalive_interval_,
364 238 : keepalive_probes_, tcp_user_timeout_);
365 : }
366 :
367 5850 : void SandeshSession::OnRead(Buffer buffer) {
368 5850 : reader_->OnRead(buffer);
369 5850 : }
370 :
371 12973 : bool SandeshSession::SendMsg(SandeshElement element) {
372 12973 : Sandesh *sandesh = element.snh_;
373 12973 : std::scoped_lock lock(send_mutex_);
374 12973 : if (!IsEstablished()) {
375 0 : if (Sandesh::IsLoggingDroppedAllowed(sandesh->type())) {
376 0 : SANDESH_LOG(ERROR, __func__ << " Not Connected : Dropping Message: " <<
377 : sandesh->ToString());
378 : }
379 0 : increment_send_msg_fail();
380 0 : Sandesh::UpdateTxMsgFailStats(sandesh->Name(), 0,
381 : SandeshTxDropReason::SessionNotConnected);
382 0 : sandesh->Release();
383 0 : return true;
384 : }
385 12973 : if (sandesh->IsLoggingAllowed()) {
386 9180 : sandesh->Log();
387 : }
388 12973 : bool more = !send_queue_->IsQueueEmpty();
389 12973 : if (stats_client_ && sandesh->type() == SandeshType::UVE) {
390 0 : stats_client_->SendMsg(sandesh);
391 : }
392 12973 : writer_->SendMsg(sandesh, more);
393 12973 : return true;
394 12973 : }
395 :
396 0 : bool SandeshSession::SendBuffer(boost::shared_ptr<TMemoryBuffer> sbuffer) {
397 0 : std::scoped_lock lock(send_mutex_);
398 0 : if (!IsEstablished()) {
399 0 : increment_send_buffer_fail();
400 0 : return true;
401 : }
402 : // No buffer packing supported currently
403 0 : writer_->SendBuffer(sbuffer);
404 0 : return true;
405 0 : }
406 :
407 0 : bool SandeshSession::EnqueueBuffer(u_int8_t *buf, u_int32_t buf_len) {
408 0 : boost::shared_ptr<TMemoryBuffer> sbuffer(new TMemoryBuffer(buf_len));
409 0 : u_int8_t *write_buf = sbuffer->getWritePtr(buf_len);
410 0 : memcpy(write_buf, buf, buf_len);
411 0 : sbuffer->wroteBytes(buf_len);
412 0 : return send_buffer_queue()->Enqueue(sbuffer);
413 0 : }
414 :
415 123 : Sandesh * SandeshSession::DecodeCtrlSandesh(const string& msg,
416 : const SandeshHeader& header,
417 : const string& sandesh_name, const uint32_t& header_offset) {
418 : namespace sandesh_prot = contrail::sandesh::protocol;
419 : namespace sandesh_trans = contrail::sandesh::transport;
420 :
421 123 : assert(header.get_Hints() & g_sandesh_constants.SANDESH_CONTROL_HINT);
422 :
423 : // Create and process the sandesh
424 123 : Sandesh *sandesh = SandeshBaseFactory::CreateInstance(sandesh_name);
425 123 : if (sandesh == NULL) {
426 0 : SANDESH_LOG(ERROR, __func__ << ": Unknown sandesh ctrl message: " << sandesh_name);
427 0 : return NULL;
428 : }
429 : boost::shared_ptr<sandesh_trans::TMemoryBuffer> btrans =
430 : boost::shared_ptr<sandesh_trans::TMemoryBuffer>(
431 123 : new sandesh_trans::TMemoryBuffer((uint8_t *)msg.c_str() + header_offset,
432 123 : msg.size() - header_offset));
433 : boost::shared_ptr<sandesh_prot::TXMLProtocol> prot =
434 123 : boost::shared_ptr<sandesh_prot::TXMLProtocol>(new sandesh_prot::TXMLProtocol(btrans));
435 123 : int32_t xfer = sandesh->Read(prot);
436 123 : if (xfer < 0) {
437 0 : SANDESH_LOG(ERROR, __func__ << ": Decoding " << sandesh_name << " for ctrl FAILED");
438 0 : sandesh->Release();
439 0 : return NULL;
440 : } else {
441 123 : return sandesh;
442 : }
443 123 : }
444 :
445 0 : void SandeshSession::EnqueueClose() {
446 0 : if (IsClosed()) {
447 0 : return;
448 : }
449 0 : std::scoped_lock lock(conn_mutex_);
450 0 : if (connection_) {
451 0 : connection_->state_machine()->OnSessionEvent(this,
452 : TcpSession::CLOSE);
453 : } else {
454 0 : TcpSession::EventObserver eobs(observer());
455 0 : if (eobs) {
456 0 : eobs(this, TcpSession::CLOSE);
457 : }
458 0 : }
459 0 : }
460 :
461 : //
462 : // SandeshReader
463 : //
464 142 : SandeshReader::SandeshReader(SandeshSession *session) :
465 142 : buf_(""),
466 142 : offset_(0),
467 142 : msg_length_(-1),
468 284 : session_(session) {
469 142 : buf_.reserve(kDefaultRecvSize);
470 142 : }
471 :
472 284 : SandeshReader::~SandeshReader() {
473 284 : }
474 :
475 123 : int SandeshReader::ExtractMsgHeader(const std::string& msg,
476 : SandeshHeader& header, std::string& msg_type, uint32_t& header_offset) {
477 123 : int32_t xfer = 0, ret;
478 : boost::shared_ptr<TMemoryBuffer> btrans =
479 : boost::shared_ptr<TMemoryBuffer>(
480 123 : new TMemoryBuffer((uint8_t *)msg.c_str(), msg.size()));
481 : boost::shared_ptr<TXMLProtocol> prot =
482 123 : boost::shared_ptr<TXMLProtocol>(new TXMLProtocol(btrans));
483 : // Read the sandesh header and note the offset
484 123 : if ((ret = header.read(prot)) <= 0) {
485 0 : SANDESH_LOG(ERROR, __func__ << ": Sandesh header read FAILED: " << msg);
486 0 : return EINVAL;
487 : }
488 123 : xfer += ret;
489 123 : header_offset = xfer;
490 : // Extract the message name
491 123 : if ((ret = prot->readSandeshBegin(msg_type)) <= 0) {
492 0 : SANDESH_LOG(ERROR, __func__ << ": Sandesh begin read FAILED: " << msg);
493 0 : return EINVAL;
494 : }
495 123 : xfer += ret;
496 123 : return 0;
497 123 : }
498 :
499 5850 : void SandeshReader::SetBuf(const std::string &str) {
500 5850 : if (buf_.empty()) {
501 4943 : ReplaceBuf(str);
502 : } else {
503 907 : buf_ += str;
504 : }
505 : // TODO handle buf_ > kMaxMessageSize
506 5850 : }
507 :
508 14282 : void SandeshReader::ReplaceBuf(const std::string &str) {
509 14282 : buf_ = str;
510 14282 : buf_.reserve(SandeshReader::kDefaultRecvSize);
511 14282 : offset_ = 0;
512 14282 : }
513 :
514 12535 : bool SandeshReader::LeftOver() const {
515 12535 : if (buf_.empty()) {
516 0 : return false;
517 : }
518 12535 : return (buf_.size() != offset_);
519 : }
520 :
521 : // Returns false if not able to extract the message length, true otherwise
522 14292 : bool SandeshReader::ExtractMsgLength(size_t &msg_length, int *result) {
523 : // Have we read enough to extract the message length?
524 14292 : if (buf_.size() - offset_ < SandeshWriter::sandesh_open_.size()) {
525 1757 : return false;
526 : }
527 : // Some sanity check
528 12535 : if (!boost::algorithm::starts_with(buf_.c_str() + offset_,
529 : SandeshWriter::sandesh_open_attr_length_)) {
530 0 : *result = -1;
531 0 : return false;
532 : }
533 :
534 12535 : std::string::const_iterator end = buf_.begin() + offset_ +
535 25070 : SandeshWriter::sandesh_open_.size() - 1;
536 12535 : if (*end != '>') {
537 0 : *result = -2;
538 0 : return false;
539 : }
540 :
541 12535 : std::string::const_iterator st = buf_.begin() + offset_ +
542 25070 : SandeshWriter::sandesh_open_attr_length_.size();
543 : // Adjust for double quote
544 12535 : --end;
545 12535 : string length = string(st, end);
546 :
547 12535 : stringToInteger(length.c_str(), msg_length);
548 12535 : if (msg_length == 0) {
549 0 : *result = -3;
550 0 : return false;
551 : }
552 12535 : return true;
553 12535 : }
554 :
555 : // Returns false if not able to extract the full message, true otherwise
556 15189 : bool SandeshReader::ExtractMsg(Buffer buffer, int *result, bool NewBuf) {
557 15189 : if (NewBuf) {
558 5850 : const uint8_t *cp = TcpSession::BufferData(buffer);
559 : // TODO Avoid this copy
560 5850 : std::string str(cp, cp + TcpSession::BufferSize(buffer));
561 5850 : SetBuf(str);
562 5850 : }
563 : // Extract the message length
564 15189 : if (!MsgLengthKnown()) {
565 14292 : size_t msg_length = 0;
566 14292 : bool done = ExtractMsgLength(msg_length, result);
567 14292 : if (done == false) {
568 1757 : return false;
569 : }
570 12535 : set_msg_length(msg_length);
571 : }
572 : // Check if the entire message is read or not
573 13432 : if (buf_.size() < msg_length()) {
574 897 : return false;
575 : }
576 12535 : return true;
577 : }
578 :
579 5850 : void SandeshReader::OnRead(Buffer buffer) {
580 5850 : std::scoped_lock lock(cb_mutex_);
581 : // Check if session is being deleted, then drop the packet
582 5850 : if (cb_.empty()) {
583 0 : SANDESH_LOG(ERROR, __func__ <<
584 : " Session being deleted: Dropping Message");
585 0 : session_->increment_recv_fail();
586 0 : session_->ReleaseBuffer(buffer);
587 0 : return;
588 : }
589 5850 : int result = 0;
590 5850 : bool done = ExtractMsg(buffer, &result, true);
591 : do {
592 15189 : if (result < 0) {
593 : // Generate error and close connection
594 0 : SANDESH_LOG(ERROR, __func__ << " Message extract failed: " << result);
595 0 : const uint8_t *cp = TcpSession::BufferData(buffer);
596 0 : size_t cp_size = TcpSession::BufferSize(buffer);
597 0 : SANDESH_LOG(ERROR, __func__ << " OnRead Buffer Size: " << cp_size);
598 0 : SANDESH_LOG(ERROR, __func__ << " OnRead Buffer: ");
599 0 : std::string debug((const char*)cp, cp_size);
600 0 : SANDESH_LOG(ERROR, debug);
601 0 : SANDESH_LOG(ERROR, __func__ << " Reader Size: " << buf_.size());
602 0 : SANDESH_LOG(ERROR, __func__ << " Reader Offset: " << offset_);
603 0 : SANDESH_LOG(ERROR, __func__ << " Reader Buffer: " << buf_);
604 0 : buf_.clear();
605 0 : offset_ = 0;
606 : // Enqueue a close on the state machine
607 0 : session_->increment_recv_fail();
608 0 : session_->EnqueueClose();
609 0 : break;
610 0 : }
611 15189 : if (done == true) {
612 : // We got good match. Process the message after extracting out
613 : // the sandesh open and close envelope
614 12535 : std::string::const_iterator st = buf_.begin() + offset_ +
615 25070 : SandeshWriter::sandesh_open_.size();
616 12535 : std::string::const_iterator end = buf_.begin() + offset_ +
617 25070 : msg_length() - SandeshWriter::sandesh_close_.size();
618 12535 : std::string xml(st, end);
619 12535 : offset_ += msg_length();
620 12535 : reset_msg_length();
621 12535 : if (!cb_(xml, session_)) {
622 : // Enqueue a close on the state machine
623 0 : session_->increment_recv_fail();
624 0 : session_->EnqueueClose();
625 0 : break;
626 : }
627 12535 : } else {
628 : // Read more data.
629 2654 : break;
630 : }
631 :
632 12535 : if (LeftOver()) {
633 9339 : ReplaceBuf(string(buf_, offset_, buf_.size() - offset_));
634 9339 : done = ExtractMsg(buffer, &result, false);
635 : } else {
636 : // No more data in the Buffer
637 3196 : buf_.clear();
638 3196 : offset_ = 0;
639 3196 : break;
640 : }
641 9339 : } while (true);
642 :
643 5850 : session_->ReleaseBuffer(buffer);
644 5850 : return;
645 5850 : }
646 :
647 284 : void SandeshReader::SetReceiveMsgCb(SandeshReceiveMsgCb cb) {
648 284 : std::scoped_lock lock(cb_mutex_);
649 284 : cb_ = cb;
650 284 : }
|
