Expand traffic monitoring to PostgreSQL and N2K outputs, handle empty AIS messages in psql

Author: jvde-github

Source/DBMS/PostgreSQL.cpp

@@ -35,19 +35,24 @@ namespace IO
 	{
 		if (PQstatus(con) != CONNECTION_OK)
 		{
+			stats.connected = 0;
 			Warning() << "DBMS: Connection to PostgreSQL lost. Attempting to reset...";
 			PQreset(con);
 
 			if (PQstatus(con) != CONNECTION_OK)
 			{
 				Error() << "DBMS: Could not reset connection. Aborting post.";
 				conn_fails++;
+				stats.connect_fail++;
 				return;
 			}
 			else
 			{
 				Warning() << "DBMS: Connection successfully reset.";
 				conn_fails = 0;
+				stats.connected = 1;
+				stats.connect_ok++;
+				stats.reconnects++;
 			}
 		}
 
@@ -92,7 +97,11 @@ namespace IO
 								   7, nullptr, params, nullptr, nullptr, 0);
 
 				if (PQresultStatus(res) == PGRES_TUPLES_OK && PQntuples(res) > 0)
+				{
 					msg_id = PQgetvalue(res, 0, 0);
+					for (const char *p : params)
+						stats.bytes_out += strlen(p);
+				}
 				else
 				{
 					Error() << "DBMS: ais_message insert failed: " << PQerrorMessage(con);
@@ -122,6 +131,10 @@ namespace IO
 						Error() << "DBMS: ais_nmea insert failed: " << PQerrorMessage(con);
 						ok = false;
 					}
+					else
+					{
+						stats.bytes_out += nmea.length();
+					}
 					PQclear(res);
 					if (!ok)
 						break;
@@ -169,8 +182,8 @@ namespace IO
 				}
 			}
 
-			// 4. Vessel upsert
-			if (ok && VD)
+			// 4. Vessel upsert (requires identifiable mmsi)
+			if (ok && VD && entry.mmsi != "0")
 				ok = execVessel(entry, msg_id_ptr);
 
 			// 5. Property inserts
@@ -375,9 +388,14 @@ namespace IO
 		con = PQconnectdb(conn_string.c_str());
 
 		if (con == nullptr || PQstatus(con) != CONNECTION_OK)
+		{
+			stats.connect_fail++;
 			throw std::runtime_error("DBMS: cannot open database :" + std::string(PQerrorMessage(con)));
+		}
 
 		conn_fails = 0;
+		stats.connected = 1;
+		stats.connect_ok++;
 
 		PGresult *res = PQexec(con, "SELECT key_id, key_str FROM ais_keys");
 		if (PQresultStatus(res) != PGRES_TUPLES_OK)

Source/DBMS/PostgreSQL.h

@@ -91,6 +91,9 @@ namespace IO
 
 #ifdef HASPSQL
 		void process();
+		using StreamIn<AIS::Message>::Receive;
+		using StreamIn<AIS::GPS>::Receive;
+		using StreamIn<JSON::JSON>::Receive;
 		void Receive(const JSON::JSON *data, int len, TAG &tag);
 #endif
 

Source/IO/N2KInterface.cpp

@@ -223,6 +223,7 @@ namespace N2K
 		{
 			std::lock_guard<std::mutex> lock(mtx);
 			NMEA2000.SendMsg(N2kMsg);
+			bytes_sent += N2kMsg.DataLen;
 		}
 	}
 

Source/IO/N2KInterface.h

@@ -78,6 +78,7 @@ namespace N2K
 		bool output = false;
 
 		bool connected = false;
+		uint64_t bytes_sent = 0;
 
 		void onOpen();
 		static void onOpenStatic();
@@ -128,6 +129,9 @@ namespace N2K
 		}
 
 		void sendMsg(const tN2kMsg &N2kMsg);
+
+		bool isConnected() const { return connected; }
+		uint64_t getBytesSent() const { return bytes_sent; }
 	};
 
 #else
@@ -136,6 +140,8 @@ namespace N2K
 	public:
 		void Start() {}
 		void Stop() {}
+		bool isConnected() const { return false; }
+		uint64_t getBytesSent() const { return 0; }
 	};
 
 #endif

Source/IO/N2KStream.cpp

@@ -811,6 +811,12 @@ namespace IO
 			default:
 				break;
 			}
+
+			bool was_connected = stats.connected != 0;
+			stats.connected = N2K::N2KInterface.isConnected() ? 1u : 0u;
+			if (!was_connected && stats.connected)
+				stats.connect_ok++;
+			stats.bytes_out = N2K::N2KInterface.getBytesSent();
 			return;
 		}
 	}