diff --git a/C/services/north/data_load.cpp b/C/services/north/data_load.cpp index 479cb48048..2fcf87559c 100755 --- a/C/services/north/data_load.cpp +++ b/C/services/north/data_load.cpp @@ -118,6 +118,8 @@ bool DataLoad::setDataSource(const string& source) source.c_str(), m_name.c_str()); return false; } + m_lastFetched = getLastSentId(); + m_streamSent = getLastSentId(); return true; } @@ -349,7 +351,7 @@ unsigned long DataLoad::getLastSentId() if (row) { // Get column value - ResultSet::ColumnValue* theVal = row->getColumn("last_object"); + ResultSet::ColumnValue* theVal = row->getColumn(getStatsColumnName()); // Set found id unsigned long rval = (unsigned long)theVal->getInteger(); delete lastObjectId; @@ -457,7 +459,7 @@ int streamId = 0; InsertValues streamValues; streamValues.push_back(InsertValue("description", m_name)); - streamValues.push_back(InsertValue("last_object", 0)); + streamValues.push_back(InsertValue(getStatsColumnName(), 0)); if (m_storage->insertTable("streams", streamValues) != 1) { @@ -515,7 +517,7 @@ void DataLoad::flushLastSentId() Where where("id", condition, to_string(m_streamId)); InsertValues lastId; - lastId.push_back(InsertValue("last_object", (long)m_streamSent)); + lastId.push_back(InsertValue(getStatsColumnName(), (long)m_streamSent)); m_storage->updateTable("streams", lastId, where); } @@ -724,3 +726,22 @@ void DataLoad::configChange(const string& category, const string& newConfig) } } } + +/** + * Get the stats column name to be used to fetch last reading sent with this current data source + */ + +std::string DataLoad::getStatsColumnName() +{ + std::string lastColName = "last_object"; + switch(m_dataSource) + { + case SourceStatistics: + lastColName = "stats_last_object"; + break; + case SourceAudit: + lastColName = "audit_last_object"; + break; + } + return lastColName; +} diff --git a/C/services/north/include/data_load.h b/C/services/north/include/data_load.h index 96d0e9f826..cec8c51934 100644 --- a/C/services/north/include/data_load.h +++ b/C/services/north/include/data_load.h @@ -70,6 +70,7 @@ class DataLoad : public ServiceHandler { ReadingSet *fetchAudit(unsigned int blockSize); void bufferReadings(ReadingSet *readings); bool loadFilters(const std::string& category); + std::string getStatsColumnName(); private: const std::string& m_name; diff --git a/VERSION b/VERSION index 4ba94eff21..f24ecd1db4 100644 --- a/VERSION +++ b/VERSION @@ -1,2 +1,2 @@ fledge_version=2.6.0 -fledge_schema=74 +fledge_schema=75 \ No newline at end of file diff --git a/scripts/plugins/storage/postgres/downgrade/74.sql b/scripts/plugins/storage/postgres/downgrade/74.sql new file mode 100644 index 0000000000..1633b44203 --- /dev/null +++ b/scripts/plugins/storage/postgres/downgrade/74.sql @@ -0,0 +1,2 @@ +ALTER TABLE fledge.streams DROP COLUMN stats_last_object; +ALTER TABLE fledge.streams DROP COLUMN audit_last_object; diff --git a/scripts/plugins/storage/postgres/init.sql b/scripts/plugins/storage/postgres/init.sql index 0337c4e872..96604a3ec2 100644 --- a/scripts/plugins/storage/postgres/init.sql +++ b/scripts/plugins/storage/postgres/init.sql @@ -423,15 +423,17 @@ CREATE INDEX readings_ix3 -- Streams table -- List of the streams to the Cloud. CREATE TABLE fledge.streams ( - id integer NOT NULL DEFAULT nextval('fledge.streams_id_seq'::regclass), -- Sequence ID - description character varying(255) NOT NULL DEFAULT ''::character varying COLLATE pg_catalog."default", -- A brief description of the stream entry - properties jsonb NOT NULL DEFAULT '{}'::jsonb, -- A generic set of properties - object_stream jsonb NOT NULL DEFAULT '{}'::jsonb, -- Definition of what must be streamed - object_block jsonb NOT NULL DEFAULT '{}'::jsonb, -- Definition of how the stream must be organised - object_filter jsonb NOT NULL DEFAULT '{}'::jsonb, -- Any filter involved in selecting the data to stream - active_window jsonb NOT NULL DEFAULT '{}'::jsonb, -- The window of operations - active boolean NOT NULL DEFAULT true, -- When false, all data to this stream stop and are inactive - last_object bigint NOT NULL DEFAULT 0, -- The ID of the last object streamed (asset or reading, depending on the object_stream) + id integer NOT NULL DEFAULT nextval('fledge.streams_id_seq'::regclass), -- Sequence ID + description character varying(255) NOT NULL DEFAULT ''::character varying COLLATE pg_catalog."default", -- A brief description of the stream entry + properties jsonb NOT NULL DEFAULT '{}'::jsonb, -- A generic set of properties + object_stream jsonb NOT NULL DEFAULT '{}'::jsonb, -- Definition of what must be streamed + object_block jsonb NOT NULL DEFAULT '{}'::jsonb, -- Definition of how the stream must be organised + object_filter jsonb NOT NULL DEFAULT '{}'::jsonb, -- Any filter involved in selecting the data to stream + active_window jsonb NOT NULL DEFAULT '{}'::jsonb, -- The window of operations + active boolean NOT NULL DEFAULT true, -- When false, all data to this stream stop and are inactive + last_object bigint NOT NULL DEFAULT 0, -- The ID of the last object streamed (asset or reading, depending on the object_stream) + stats_last_object bigint NOT NULL DEFAULT 0, -- The ID of the last object streamed (asset or reading, depending on the object_stream) + audit_last_object bigint NOT NULL DEFAULT 0, -- The ID of the last object streamed (asset or reading, depending on the object_stream) ts timestamp(6) with time zone NOT NULL DEFAULT now(), -- Creation or last update CONSTRAINT strerams_pkey PRIMARY KEY (id)); diff --git a/scripts/plugins/storage/postgres/upgrade/75.sql b/scripts/plugins/storage/postgres/upgrade/75.sql new file mode 100644 index 0000000000..b97295ebe2 --- /dev/null +++ b/scripts/plugins/storage/postgres/upgrade/75.sql @@ -0,0 +1,2 @@ +ALTER TABLE fledge.streams ADD COLUMN stats_last_object bigint DEFAULT 0; +ALTER TABLE fledge.streams ADD COLUMN audit_last_object bigint DEFAULT 0; diff --git a/scripts/plugins/storage/sqlite/downgrade/74.sql b/scripts/plugins/storage/sqlite/downgrade/74.sql new file mode 100644 index 0000000000..1633b44203 --- /dev/null +++ b/scripts/plugins/storage/sqlite/downgrade/74.sql @@ -0,0 +1,2 @@ +ALTER TABLE fledge.streams DROP COLUMN stats_last_object; +ALTER TABLE fledge.streams DROP COLUMN audit_last_object; diff --git a/scripts/plugins/storage/sqlite/init.sql b/scripts/plugins/storage/sqlite/init.sql index cb35730643..552ab25ae9 100644 --- a/scripts/plugins/storage/sqlite/init.sql +++ b/scripts/plugins/storage/sqlite/init.sql @@ -212,15 +212,17 @@ CREATE INDEX fki_asset_messages_fk2 -- Streams table -- List of the streams to the Cloud. CREATE TABLE fledge.streams ( - id INTEGER PRIMARY KEY AUTOINCREMENT, -- Sequence ID - description character varying(255) NOT NULL DEFAULT '', -- A brief description of the stream entry - properties JSON NOT NULL DEFAULT '{}', -- A generic set of properties - object_stream JSON NOT NULL DEFAULT '{}', -- Definition of what must be streamed - object_block JSON NOT NULL DEFAULT '{}', -- Definition of how the stream must be organised - object_filter JSON NOT NULL DEFAULT '{}', -- Any filter involved in selecting the data to stream - active_window JSON NOT NULL DEFAULT '{}', -- The window of operations - active boolean NOT NULL DEFAULT 't', -- When false, all data to this stream stop and are inactive - last_object bigint NOT NULL DEFAULT 0, -- The ID of the last object streamed (asset or reading, depending on the object_stream) + id INTEGER PRIMARY KEY AUTOINCREMENT, -- Sequence ID + description character varying(255) NOT NULL DEFAULT '', -- A brief description of the stream entry + properties JSON NOT NULL DEFAULT '{}', -- A generic set of properties + object_stream JSON NOT NULL DEFAULT '{}', -- Definition of what must be streamed + object_block JSON NOT NULL DEFAULT '{}', -- Definition of how the stream must be organised + object_filter JSON NOT NULL DEFAULT '{}', -- Any filter involved in selecting the data to stream + active_window JSON NOT NULL DEFAULT '{}', -- The window of operations + active boolean NOT NULL DEFAULT 't', -- When false, all data to this stream stop and are inactive + last_object bigint NOT NULL DEFAULT 0, -- The ID of the last object streamed (asset or reading, depending on the object_stream) + stats_last_object bigint NOT NULL DEFAULT 0, -- The ID of the last object streamed (Statistics, depending on the object_stream) + audit_last_object bigint NOT NULL DEFAULT 0, -- The ID of the last object streamed (Audit, depending on the object_stream) ts DATETIME DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW', 'localtime'))); -- Creation or last update diff --git a/scripts/plugins/storage/sqlite/upgrade/75.sql b/scripts/plugins/storage/sqlite/upgrade/75.sql new file mode 100644 index 0000000000..b97295ebe2 --- /dev/null +++ b/scripts/plugins/storage/sqlite/upgrade/75.sql @@ -0,0 +1,2 @@ +ALTER TABLE fledge.streams ADD COLUMN stats_last_object bigint DEFAULT 0; +ALTER TABLE fledge.streams ADD COLUMN audit_last_object bigint DEFAULT 0; diff --git a/scripts/plugins/storage/sqlitelb/downgrade/74.sql b/scripts/plugins/storage/sqlitelb/downgrade/74.sql new file mode 100644 index 0000000000..1633b44203 --- /dev/null +++ b/scripts/plugins/storage/sqlitelb/downgrade/74.sql @@ -0,0 +1,2 @@ +ALTER TABLE fledge.streams DROP COLUMN stats_last_object; +ALTER TABLE fledge.streams DROP COLUMN audit_last_object; diff --git a/scripts/plugins/storage/sqlitelb/init.sql b/scripts/plugins/storage/sqlitelb/init.sql index 48b7771f85..30cc00b127 100644 --- a/scripts/plugins/storage/sqlitelb/init.sql +++ b/scripts/plugins/storage/sqlitelb/init.sql @@ -212,15 +212,17 @@ CREATE INDEX fki_asset_messages_fk2 -- Streams table -- List of the streams to the Cloud. CREATE TABLE fledge.streams ( - id INTEGER PRIMARY KEY AUTOINCREMENT, -- Sequence ID - description character varying(255) NOT NULL DEFAULT '', -- A brief description of the stream entry - properties JSON NOT NULL DEFAULT '{}', -- A generic set of properties - object_stream JSON NOT NULL DEFAULT '{}', -- Definition of what must be streamed - object_block JSON NOT NULL DEFAULT '{}', -- Definition of how the stream must be organised - object_filter JSON NOT NULL DEFAULT '{}', -- Any filter involved in selecting the data to stream - active_window JSON NOT NULL DEFAULT '{}', -- The window of operations - active boolean NOT NULL DEFAULT 't', -- When false, all data to this stream stop and are inactive - last_object bigint NOT NULL DEFAULT 0, -- The ID of the last object streamed (asset or reading, depending on the object_stream) + id INTEGER PRIMARY KEY AUTOINCREMENT, -- Sequence ID + description character varying(255) NOT NULL DEFAULT '', -- A brief description of the stream entry + properties JSON NOT NULL DEFAULT '{}', -- A generic set of properties + object_stream JSON NOT NULL DEFAULT '{}', -- Definition of what must be streamed + object_block JSON NOT NULL DEFAULT '{}', -- Definition of how the stream must be organised + object_filter JSON NOT NULL DEFAULT '{}', -- Any filter involved in selecting the data to stream + active_window JSON NOT NULL DEFAULT '{}', -- The window of operations + active boolean NOT NULL DEFAULT 't', -- When false, all data to this stream stop and are inactive + last_object bigint NOT NULL DEFAULT 0, -- The ID of the last object streamed (asset or reading, depending on the object_stream) + stats_last_object bigint NOT NULL DEFAULT 0, -- The ID of the last object streamed (Statistics, depending on the object_stream) + audit_last_object bigint NOT NULL DEFAULT 0, -- The ID of the last object streamed (Audit, depending on the object_stream) ts DATETIME DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW', 'localtime'))); -- Creation or last update diff --git a/scripts/plugins/storage/sqlitelb/upgrade/75.sql b/scripts/plugins/storage/sqlitelb/upgrade/75.sql new file mode 100644 index 0000000000..b97295ebe2 --- /dev/null +++ b/scripts/plugins/storage/sqlitelb/upgrade/75.sql @@ -0,0 +1,2 @@ +ALTER TABLE fledge.streams ADD COLUMN stats_last_object bigint DEFAULT 0; +ALTER TABLE fledge.streams ADD COLUMN audit_last_object bigint DEFAULT 0;