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