| 427 | } |
| 428 | |
| 429 | void InfluxdbCommonWriter::FlushWQ() |
| 430 | { |
| 431 | AssertOnWorkQueue(); |
| 432 | |
| 433 | namespace beast = boost::beast; |
| 434 | namespace http = beast::http; |
| 435 | |
| 436 | /* Flush can be called from 1) Timeout 2) Threshold 3) on shutdown/reload. */ |
| 437 | if (m_DataBuffer.empty()) |
| 438 | return; |
| 439 | |
| 440 | Log(LogDebug, GetReflectionType()->GetName()) |
| 441 | << "Flushing data buffer to InfluxDB."; |
| 442 | |
| 443 | String body = boost::algorithm::join(m_DataBuffer, "\n"); |
| 444 | m_DataBuffer.clear(); |
| 445 | m_DataBufferSize = 0; |
| 446 | |
| 447 | OptionalTlsStream stream; |
| 448 | |
| 449 | try { |
| 450 | stream = Connect(); |
| 451 | } catch (const std::exception& ex) { |
| 452 | Log(LogWarning, GetReflectionType()->GetName()) |
| 453 | << "Flush failed, cannot connect to InfluxDB: " << DiagnosticInformation(ex, false); |
| 454 | return; |
| 455 | } |
| 456 | |
| 457 | Defer s ([&stream]() { |
| 458 | if (stream.first) { |
| 459 | stream.first->next_layer().shutdown(); |
| 460 | } |
| 461 | }); |
| 462 | |
| 463 | auto request (AssembleRequest(std::move(body))); |
| 464 | |
| 465 | try { |
| 466 | if (stream.first) { |
| 467 | http::write(*stream.first, request); |
| 468 | stream.first->flush(); |
| 469 | } else { |
| 470 | http::write(*stream.second, request); |
| 471 | stream.second->flush(); |
| 472 | } |
| 473 | } catch (const std::exception& ex) { |
| 474 | Log(LogWarning, GetReflectionType()->GetName()) |
| 475 | << "Cannot write to TCP socket on host '" << GetHost() << "' port '" << GetPort() << "'."; |
| 476 | throw; |
| 477 | } |
| 478 | |
| 479 | http::parser<false, http::string_body> parser; |
| 480 | beast::flat_buffer buf; |
| 481 | |
| 482 | try { |
| 483 | if (stream.first) { |
| 484 | http::read(*stream.first, buf, parser); |
| 485 | } else { |
| 486 | http::read(*stream.second, buf, parser); |