mirror/thingsboard-gateway
1
0
Fork 0
mirror of https://github.com/thingsboard/thingsboard-gateway synced 2025-10-26 22:31:42 +08:00
Code Issues Releases Wiki Activity

Fixed sqlite storage

Browse Source
This commit is contained in:
samson0v
2021-09-28 16:22:49 +03:00
parent daa865d644
commit 5b021029a2
25 changed files with 237 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
.gitignore vendored
View File

@@ -5,9 +5,6 @@ thingsboard_gateway/develop_tests/
thingsboard_gateway/logs/*
thingsboard_gateway/config/connected_devices.json
thingsboard_gateway/storage/data/
thingsboard_gateway.egg-info/
.vscode/
/debug_utils/
.idea/
/build/
/deb_dist/
@@ -15,14 +12,13 @@ thingsboard_gateway.egg-info/
/.mypy_cache/
/thingsboard_gateway/storage/data_out/
*.gz
*.orig
/logs/
/python3-thingsboard-gateway.deb
/python3-thingsboard-gateway.rpm
/thingsboard_gateway.egg-info/
/docker/python3-thingsboard-gateway.deb
/data/
/tests/storage/data/
/logs/
__pycache__
/thingsboard_gateway/config/
/venv/
*.db

3
for_build/DEBIAN/postinst
View File

@@ -1,4 +1,3 @@
#!/bin/sh
# Copyright 2021. ThingsBoard
#
@@ -46,4 +45,4 @@ echo "Enabling daemon..."
sudo pidof systemd && sudo systemctl enable thingsboard-gateway || echo "Systemctl not found"
#echo "Daemon starting..."
sudo pidof systemd && sudo systemctl start thingsboard-gateway || echo
echo -e "\e[96mThingsboard Gateway \e[92mhas been installed. Have a nice day \e[93m\e[5m:)\e[25m\e[39m"
echo -e "\e[96mThingsboard Gateway \e[92mhas been installed. Have a nice day \e[93m\e[5m:)\e[25m\e[39m"

2
for_build/DEBIAN/preinst
View File

@@ -18,4 +18,4 @@ set -e
echo "Installing directory for configs..."
sudo mkdir /etc/thingsboard-gateway || echo
sudo adduser --system --gecos "ThingsBoard-Gateway Service" --disabled-password --group --home /var/lib/thingsboard_gateway thingsboard_gateway || echo "User exists"
sudo mkdir /var/lib/thingsboard_gateway/extensions || echo
sudo mkdir /var/lib/thingsboard_gateway/extensions || echo

100
for_build/etc/thingsboard-gateway/config/ble.json
View File

@@ -1,53 +1,53 @@
{
"name": "BLE Connector",
"rescanIntervalSeconds": 100,
"checkIntervalSeconds": 100,
"scanTimeSeconds": 5,
"passiveScanMode": true,
"devices": [
{
"name": "Temperature and humidity sensor",
"MACAddress": "4C:65:A8:DF:85:C0",
"addrType": "public",
"telemetry": [
"name": "BLE Connector",
"rescanIntervalSeconds": 100,
"checkIntervalSeconds": 100,
"scanTimeSeconds": 5,
"passiveScanMode": true,
"devices": [
{
"key": "temperature",
"method": "notify",
"characteristicUUID": "226CAA55-6476-4566-7562-66734470666D",
"byteFrom": 2,
"byteTo": 6
},
{
"key": "humidity",
"method": "notify",
"characteristicUUID": "226CAA55-6476-4566-7562-66734470666D",
"byteFrom": 9,
"byteTo": 13
"name": "Temperature and humidity sensor",
"MACAddress": "4C:65:A8:DF:85:C0",
"addrType": "public",
"telemetry": [
{
"key": "temperature",
"method": "notify",
"characteristicUUID": "226CAA55-6476-4566-7562-66734470666D",
"byteFrom": 2,
"byteTo": 6
},
{
"key": "humidity",
"method": "notify",
"characteristicUUID": "226CAA55-6476-4566-7562-66734470666D",
"byteFrom": 9,
"byteTo": 13
}
],
"attributes": [
{
"key": "name",
"characteristicUUID": "00002A00-0000-1000-8000-00805F9B34FB",
"method": "read",
"byteFrom": 0,
"byteTo": -1
}
],
"attributeUpdates": [
{
"attributeOnThingsBoard": "sharedName",
"characteristicUUID": "00002A00-0000-1000-8000-00805F9B34FB"
}
],
"serverSideRpc": [
{
"methodRPC": "rpcMethod1",
"withResponse": true,
"characteristicUUID": "00002A00-0000-1000-8000-00805F9B34FB",
"methodProcessing": "read"
}
]
}
],
"attributes": [
{
"key": "name",
"characteristicUUID": "00002A00-0000-1000-8000-00805F9B34FB",
"method": "read",
"byteFrom": 0,
"byteTo": -1
}
],
"attributeUpdates": [
{
"attributeOnThingsBoard": "sharedName",
"characteristicUUID": "00002A00-0000-1000-8000-00805F9B34FB"
}
],
"serverSideRpc": [
{
"methodRPC": "rpcMethod1",
"withResponse": true,
"characteristicUUID": "00002A00-0000-1000-8000-00805F9B34FB",
"methodProcessing": "read"
}
]
}
]
}
]
}

4
for_build/etc/thingsboard-gateway/config/custom_serial.json
View File

@@ -14,7 +14,7 @@
"untilDelimiter": "\r"
}
],
"attributes": [
"attributes":[
{
"key": "SerialNumber",
"type": "string",
@@ -30,4 +30,4 @@
]
}
]
}
}

34
for_build/etc/thingsboard-gateway/config/logs.conf
View File

@@ -5,83 +5,83 @@ keys=consoleHandler, serviceHandler, connectorHandler, converterHandler, tb_conn
[formatters]
keys=LogFormatter
[logger_root]
level=INFO
level=ERROR
handlers=consoleHandler
[logger_connector]
level=DEBUG
level=INFO
handlers=connectorHandler
formatter=LogFormatter
qualname=connector
[logger_storage]
level=DEBUG
level=INFO
handlers=storageHandler
formatter=LogFormatter
qualname=storage
[logger_database]
level=DEBUG
level=INFO
handlers=databaseHandler
formatter=LogFormatter
qualname=database
[logger_tb_connection]
level=DEBUG
level=INFO
handlers=tb_connectionHandler
formatter=LogFormatter
qualname=tb_connection
[logger_service]
level=DEBUG
level=INFO
handlers=serviceHandler
formatter=LogFormatter
qualname=service
[logger_converter]
level=DEBUG
level=INFO
handlers=converterHandler
formatter=LogFormatter
qualname=converter
[logger_extension]
level=DEBUG
level=INFO
handlers=connectorHandler
formatter=LogFormatter
qualname=extension
[handler_consoleHandler]
class=StreamHandler
level=DEBUG
level=INFO
formatter=LogFormatter
args=(sys.stdout,)
[handler_connectorHandler]
level=DEBUG
level=INFO
class=logging.handlers.TimedRotatingFileHandler
formatter=LogFormatter
args=("./logs/connector.log", "d", 1, 7,)
[handler_storageHandler]
level=DEBUG
level=INFO
class=logging.handlers.TimedRotatingFileHandler
formatter=LogFormatter
args=("./logs/storage.log", "d", 1, 7,)
[handler_databaseHandler]
level=DEBUG
level=INFO
class=logging.handlers.TimedRotatingFileHandler
formatter=LogFormatter
args=("./logs/database.log", "d", 1, 7,)
[handler_serviceHandler]
level=DEBUG
level=INFO
class=logging.handlers.TimedRotatingFileHandler
formatter=LogFormatter
args=("./logs/service.log", "d", 1, 7,)
[handler_converterHandler]
level=DEBUG
level=INFO
class=logging.handlers.TimedRotatingFileHandler
formatter=LogFormatter
args=("./logs/converter.log", "d", 1, 3,)
[handler_extensionHandler]
level=DEBUG
level=INFO
class=logging.handlers.TimedRotatingFileHandler
formatter=LogFormatter
args=("./logs/extension.log", "d", 1, 3,)
[handler_tb_connectionHandler]
level=DEBUG
level=INFO
class=logging.handlers.TimedRotatingFileHandler
formatter=LogFormatter
args=("./logs/tb_connection.log", "d", 1, 3,)
[formatter_LogFormatter]
format="%(asctime)s - |%(levelname)s| - [%(filename)s] - %(module)s - %(funcName)s - %(lineno)d - %(message)s"
format="%(asctime)s - |%(levelname)s| - [%(filename)s] - %(module)s - %(funcName)s - %(lineno)d - %(message)s"
datefmt="%Y-%m-%d %H:%M:%S"

6
for_build/etc/thingsboard-gateway/config/modbus.json
View File

@@ -75,8 +75,8 @@
"objectsCount": 4,
"address": 13
}
],
"timeseries": [
],
"timeseries": [
{
"tag": "8uint_read",
"type": "8uint",
@@ -159,7 +159,7 @@
"address": 33
},
{
"tag": "getCPULoad",
"tag":"getCPULoad",
"type": "32int",
"functionCode": 4,
"objectsCount": 2,

10
for_build/etc/thingsboard-gateway/config/mqtt.json
View File

@@ -1,8 +1,9 @@
{
"broker": {
"name":"Default Local Broker",
"host":"192.168.1.100",
"host":"127.0.0.1",
"port":1883,
"clientId": "ThingsBoard_gateway",
"security": {
"type": "basic",
"username": "user",
@@ -22,6 +23,11 @@
"type": "string",
"key": "model",
"value": "${sensorModel}"
},
{
"type": "string",
"key": "${sensorModel}",
"value": "on"
}
],
"timeseries": [
@@ -123,4 +129,4 @@
"valueExpression": "${params}"
}
]
}
}

8
for_build/etc/thingsboard-gateway/config/odbc.json
View File

@@ -47,12 +47,8 @@
"procedureOne",
{
"name": "procedureTwo",
"args": [
"One",
2,
3.0
]
"args": [ "One", 2, 3.0 ]
}
]
}
}
}

9
for_build/etc/thingsboard-gateway/config/opcua.json
View File

@@ -4,7 +4,7 @@
"url": "localhost:4840/freeopcua/server/",
"timeoutInMillis": 5000,
"scanPeriodInMillis": 5000,
"disableSubscriptions": false,
"disableSubscriptions":false,
"subCheckPeriodInMillis": 100,
"showMap": false,
"security": "Basic128Rsa15",
@@ -34,10 +34,7 @@
"rpc_methods": [
{
"method": "multiply",
"arguments": [
2,
4
]
"arguments": [2, 4]
}
],
"attributes_updates": [
@@ -49,4 +46,4 @@
}
]
}
}
}

28
for_build/etc/thingsboard-gateway/config/request.json
View File

@@ -121,19 +121,19 @@
}
],
"attributeUpdates": [
{
"httpMethod": "POST",
"httpHeaders": {
"CONTENT-TYPE": "application/json"
},
"timeout": 0.5,
"tries": 3,
"allowRedirects": true,
"deviceNameFilter": "SD.*",
"attributeFilter": "send_data",
"requestUrlExpression": "sensor/${deviceName}/${attributeKey}",
"valueExpression": "{\"${attributeKey}\":\"${attributeValue}\"}"
}
{
"httpMethod": "POST",
"httpHeaders": {
"CONTENT-TYPE": "application/json"
},
"timeout": 0.5,
"tries": 3,
"allowRedirects": true,
"deviceNameFilter": "SD.*",
"attributeFilter": "send_data",
"requestUrlExpression": "sensor/${deviceName}/${attributeKey}",
"valueExpression": "{\"${attributeKey}\":\"${attributeValue}\"}"
}
],
"serverSideRpc": [
{
@@ -160,4 +160,4 @@
}
}
]
}
}

68
for_build/etc/thingsboard-gateway/config/rest.json
View File

@@ -1,13 +1,14 @@
{
"host": "127.0.0.1",
"port": "5000",
"mapping": [
"mapping":[
{
"endpoint": "/device1",
"HTTPMethods": [
"POST"
],
"security": {
"security":
{
"type": "basic",
"username": "user",
"password": "passwd"
@@ -43,7 +44,8 @@
"GET",
"POST"
],
"security": {
"security":
{
"type": "anonymous"
},
"converter": {
@@ -76,7 +78,8 @@
"HTTPMethods": [
"POST"
],
"security": {
"security":
{
"type": "anonymous"
},
"converter": {
@@ -86,38 +89,37 @@
"extension": "CustomRestUplinkConverter",
"extension-config": [
{
"key": "Totaliser",
"datatype": "float",
"fromByte": 0,
"toByte": 4,
"byteorder": "big",
"signed": true,
"multiplier": 1
}
]
"key": "Totaliser",
"datatype": "float",
"fromByte": 0,
"toByte": 4,
"byteorder": "big",
"signed": true,
"multiplier": 1
}]
}
}
],
"attributeUpdates": [
{
"HTTPMethod": "POST",
"SSLVerify": false,
"httpHeaders": {
"CONTENT-TYPE": "application/json"
},
"security": {
"type": "basic",
"username": "user",
"password": "passwd"
},
"timeout": 0.5,
"tries": 3,
"allowRedirects": true,
"deviceNameFilter": ".*REST$",
"attributeFilter": "data",
"requestUrlExpression": "sensor/${deviceName}/${attributeKey}",
"valueExpression": "{\"${attributeKey}\":\"${attributeValue}\"}"
}
{
"HTTPMethod": "POST",
"SSLVerify": false,
"httpHeaders": {
"CONTENT-TYPE": "application/json"
},
"security": {
"type": "basic",
"username": "user",
"password": "passwd"
},
"timeout": 0.5,
"tries": 3,
"allowRedirects": true,
"deviceNameFilter": ".*REST$",
"attributeFilter": "data",
"requestUrlExpression": "sensor/${deviceName}/${attributeKey}",
"valueExpression": "{\"${attributeKey}\":\"${attributeValue}\"}"
}
],
"serverSideRpc": [
{
@@ -147,4 +149,4 @@
}
}
]
}
}

1
for_build/etc/thingsboard-gateway/extensions/__init__.py
View File

@@ -11,3 +11,4 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

1
for_build/etc/thingsboard-gateway/extensions/ble/__init__.py
View File

@@ -11,3 +11,4 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

1
for_build/etc/thingsboard-gateway/extensions/modbus/__init__.py
View File

@@ -11,3 +11,4 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

1
for_build/etc/thingsboard-gateway/extensions/mqtt/__init__.py
View File

@@ -11,3 +11,4 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

6
for_build/etc/thingsboard-gateway/extensions/mqtt/custom_mqtt_uplink_converter.py
View File

@@ -13,7 +13,6 @@
# limitations under the License.
from simplejson import dumps
from thingsboard_gateway.connectors.mqtt.mqtt_uplink_converter import MqttUplinkConverter, log
@@ -24,8 +23,7 @@ class CustomMqttUplinkConverter(MqttUplinkConverter):
def convert(self, topic, body):
try:
self.dict_result["deviceName"] = topic.split("/")[
-1] # getting all data after last '/' symbol in this case: if topic = 'devices/temperature/sensor1' device name will be 'sensor1'.
self.dict_result["deviceName"] = topic.split("/")[-1] # getting all data after last '/' symbol in this case: if topic = 'devices/temperature/sensor1' device name will be 'sensor1'.
self.dict_result["deviceType"] = "Thermostat" # just hardcode this
self.dict_result["telemetry"] = [] # template for telemetry array
bytes_to_read = body.replace("0x", "") # Replacing the 0x (if '0x' in body), needs for converting to bytearray
@@ -34,7 +32,7 @@ class CustomMqttUplinkConverter(MqttUplinkConverter):
for telemetry_key in self.__config["extension-config"]: # Processing every telemetry key in config for extension
value = 0
for _ in range(self.__config["extension-config"][telemetry_key]): # reading every value with value length from config
value = value * 256 + converted_bytes.pop(0) # process and remove byte from processing
value = value*256 + converted_bytes.pop(0) # process and remove byte from processing
telemetry_to_send = {telemetry_key.replace("Bytes", ""): value} # creating telemetry data for sending into Thingsboard
self.dict_result["telemetry"].append(telemetry_to_send) # adding data to telemetry array
else:

1
for_build/etc/thingsboard-gateway/extensions/opcua/__init__.py
View File

@@ -11,3 +11,4 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

2
for_build/etc/thingsboard-gateway/extensions/request/custom_request_uplink_converter.py
View File

@@ -16,8 +16,8 @@ import struct
from simplejson import dumps
from thingsboard_gateway.connectors.request.request_converter import RequestConverter, log
from thingsboard_gateway.tb_utility.tb_utility import TBUtility
from thingsboard_gateway.connectors.request.request_converter import RequestConverter, log
class CustomRequestUplinkConverter(RequestConverter):

2
for_build/etc/thingsboard-gateway/extensions/serial/custom_serial_converter.py
View File

@@ -23,7 +23,7 @@ class CustomSerialUplinkConverter(Converter):
'deviceType': config.get('deviceType', 'default'),
'attributes': [],
'telemetry': []
}
}
def convert(self, config, data: bytes):
keys = ['attributes', 'telemetry']

2
setup.py
View File

@@ -34,7 +34,7 @@ setup(
long_description=long_description,
long_description_content_type="text/markdown",
include_package_data=True,
python_requires=">=3.5",
python_requires=">=3.7",
packages=['thingsboard_gateway', 'thingsboard_gateway.gateway', 'thingsboard_gateway.storage','thingsboard_gateway.storage.memory',
'thingsboard_gateway.storage.file','thingsboard_gateway.storage.sqlite','thingsboard_gateway.tb_client',
'thingsboard_gateway.connectors', 'thingsboard_gateway.connectors.ble',

59
tests/tests.py
View File

@@ -24,7 +24,8 @@ from thingsboard_gateway.connectors.modbus.bytes_modbus_uplink_converter import
from thingsboard_gateway.connectors.opcua.opcua_uplink_converter import OpcUaUplinkConverter
from thingsboard_gateway.connectors.ble.bytes_ble_uplink_converter import BytesBLEUplinkConverter
from thingsboard_gateway.connectors.request.json_request_uplink_converter import JsonRequestUplinkConverter
from thingsboard_gateway.storage.memory_event_storage import MemoryEventStorage
from thingsboard_gateway.storage.file_event_storage import FileEventStorage
logging.basicConfig(level=logging.ERROR,
format='%(asctime)s - %(levelname)s - %(module)s - %(lineno)d - %(message)s',
@@ -384,5 +385,61 @@ class ConvertersTests(unittest.TestCase):
self.assertDictEqual(result, test_request_result)
class TestStorage(unittest.TestCase):
def test_memory_storage(self):
test_size = randint(0, 100)
storage_test_config = {
"type": "memory",
"read_records_count": 10,
"max_records_count": test_size * 10
}
storage = MemoryEventStorage(storage_test_config)
for test_value in range(test_size * 10):
storage.put(test_value)
result = []
for _ in range(test_size):
result.append(storage.get_event_pack())
storage.event_pack_processing_done()
correct_result = [[item for item in range(pack * 10, (pack + 1) * 10)] for pack in range(test_size)]
self.assertListEqual(result, correct_result)
def test_file_storage(self):
storage_test_config = {"data_folder_path": "storage/data/",
"max_file_count": 1000,
"max_records_per_file": 10,
"max_read_records_count": 10,
"no_records_sleep_interval": 5000
}
test_size = randint(0, storage_test_config["max_file_count"]-1)
storage = FileEventStorage(storage_test_config)
for test_value in range(test_size * 10):
storage.put(str(test_value))
sleep(.01)
result = []
for _ in range(test_size):
batch = storage.get_event_pack()
result.append(batch)
storage.event_pack_processing_done()
correct_result = [[str(x) for x in range(y * 10, (y + 1) * 10)] for y in range(test_size)]
print(result)
print(correct_result)
for file in listdir(storage_test_config["data_folder_path"]):
remove(storage_test_config["data_folder_path"]+"/"+file)
removedirs(storage_test_config["data_folder_path"])
self.assertListEqual(result, correct_result)
if __name__ == '__main__':
unittest.main()

36
thingsboard_gateway/gateway/tb_gateway_service.py
View File

@@ -37,8 +37,6 @@ from thingsboard_gateway.tb_utility.tb_updater import TBUpdater
from thingsboard_gateway.tb_utility.tb_utility import TBUtility
from thingsboard_gateway.storage.sqlite.storage_handler import StorageHandler
from thingsboard_gateway.storage.sqlite.database_action_type import DatabaseActionType
from thingsboard_gateway.storage.sqlite.database_request import DatabaseRequest
log = logging.getLogger('service')
main_handler = logging.handlers.MemoryHandler(-1)
@@ -55,7 +53,7 @@ DEFAULT_CONNECTORS = {
"rest": "RESTConnector",
"snmp": "SNMPConnector",
"ftp": "FTPConnector"
}
}
class TBGatewayService:
@@ -136,13 +134,8 @@ class TBGatewayService:
self.connectors_configs = {}
self.__remote_configurator = None
self.__request_config_after_connect = False
# if self.__config['storage']['type'] == 'sqlite':
# self.__connected_devices = self._event_storage.connected_devices
# else:
self.__connected_devices = {}
# self.__load_persistent_devices()
self.__load_persistent_devices()
self.__init_remote_configuration()
self._load_connectors()
self._connect_with_connectors()
@@ -244,10 +237,7 @@ class TBGatewayService:
self.__updater.stop()
log.info("Stopping...")
self.__close_connectors()
if self.__config['storage']['type'] == 'sqlite':
self._event_storage.close_db()
self._event_storage.stop()
log.info("The gateway has been stopped.")
self.tb_client.disconnect()
self.tb_client.stop()
@@ -415,6 +405,7 @@ class TBGatewayService:
data["telemetry"] = telemetry_with_ts
else:
data["telemetry"] = {"ts": int(time() * 1000), "values": telemetry}
json_data = dumps(data)
save_result = self._event_storage.put(json_data)
if not save_result:
@@ -434,9 +425,6 @@ class TBGatewayService:
devices_data_in_event_pack = {}
log.debug("Send data Thread has been started successfully.")
# disconnected = False
# last_disconnect = None
while not self.stopped:
try:
if self.tb_client.is_connected():
@@ -630,17 +618,6 @@ class TBGatewayService:
def __rpc_ping(*args):
return {"code": 200, "resp": "pong"}
# Request data from gateway storage
# you have to specify timestamp to specify from when to start reading
# args: JSON{"deviceName": "<deviceName>", "ts": <timestamp>}
def __rpc_data(self, args):
arguments = loads(args)
log.debug(str(arguments))
_type = DatabaseActionType.READ_DEVICE
request = DatabaseRequest(_type, arguments)
self._event_storage.processQueue.put(request)
return {"code": 200, "resp": "Data from %s scheduled for upload" % arguments.get("deviceName")}
def __rpc_devices(self, *args):
data_to_send = {}
for device in self.__connected_devices:
@@ -747,6 +724,11 @@ class TBGatewayService:
self.__save_persistent_devices()
self.tb_client.client.gw_connect_device(device_name, device_type)
def update_device(self, device_name, event, content):
if event == 'connector' and self.__connected_devices[device_name].get(event) != content:
self.__save_persistent_devices()
self.__connected_devices[device_name][event] = content
def del_device(self, device_name):
del self.__connected_devices[device_name]
del self.__saved_devices[device_name]

7
thingsboard_gateway/storage/file/event_storage_reader.py
View File

@@ -106,13 +106,6 @@ class EventStorageReader:
try:
if self.buffered_reader is None or self.buffered_reader.closed:
new_file_to_read_path = self.settings.get_data_folder_path() + pointer.get_file()
# if not exists(new_file_to_read_path):
# next_file = self.get_next_file(self.files, self.new_pos)
# if next_file is not None:
# new_file_to_read_path = self.settings.get_data_folder_path() + next_file
# else:
# self.buffered_reader = None
# return None
self.buffered_reader = BufferedReader(FileIO(new_file_to_read_path, 'r'))
lines_to_skip = pointer.get_line()
if lines_to_skip > 0:

37
thingsboard_gateway/storage/sqlite/storage_handler.py
View File

@@ -11,7 +11,7 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import sqlite3
from time import time
from thingsboard_gateway.storage.event_storage import EventStorage
@@ -47,17 +47,21 @@ class StorageHandler(EventStorage):
log.info("Sqlite storage initialized!")
self.delete_time_point = None
self.last_read = time()
self.closed = False
self.stopped = False
def get_event_pack(self):
self.delete_time_point = self.last_read
data_from_storage = self.read_data(self.last_read)
self.last_read = time()
if not self.stopped:
self.delete_time_point = self.last_read
data_from_storage = self.read_data(self.last_read)
self.last_read = time()
return [item[0] for item in data_from_storage or []]
return [item[0] for item in data_from_storage or []]
else:
return []
def event_pack_processing_done(self):
self.delete_data(self.delete_time_point)
if not self.stopped:
self.delete_data(self.delete_time_point)
def read_data(self, ts):
return self.db.read_data(ts)
@@ -67,16 +71,19 @@ class StorageHandler(EventStorage):
def put(self, message):
try:
_type = DatabaseActionType.WRITE_DATA_STORAGE
request = DatabaseRequest(_type, message)
if not self.stopped:
_type = DatabaseActionType.WRITE_DATA_STORAGE
request = DatabaseRequest(_type, message)
log.info("Sending data to storage")
self.processQueue.put(request)
self.db.process()
return True
log.info("Sending data to storage")
self.processQueue.put(request)
self.db.process()
return True
else:
return False
except Exception as e:
log.exception(e)
def close_db(self):
def stop(self):
self.db.closeDB()
self.closed = True
self.stopped = True