/* This Source Code Form is subject to the terms of the Mozilla Public * License, v. 2.0. If a copy of the MPL was not distributed with this * file, You can obtain one at http://mozilla.org/MPL/2.0/. * * Copyright 2018 (c) basysKom GmbH (Author: Peter Rustler) */ #pragma GCC diagnostic push // require GCC 4.6 #pragma GCC diagnostic ignored "-Wcast-qual" #pragma GCC diagnostic ignored "-Wsign-conversion" #include #include #include #include #include #include /**************************************************************************************************************************/ static int hash_check_args(void * h_ptr, const void * h_strkey, unsigned int h_strlen, const char * func) { if (h_ptr == NULL || h_strkey == NULL || h_strlen == 0 || h_strlen >= HASH_INTPTR_MAXLEN) { fprintf(stderr, "Error, invalid arguments in [%s]: %p, %p, %d\n", func, h_ptr, h_strkey, h_strlen); fflush(stderr); return HASH_INTPTR_ERROR_ARGS; } return 0; } int hash_intptr_addintptr(struct hash_intptr * * p_hash_intptr, const void * p_strkey, unsigned int p_strlen,int p_int_value, void * p_ptr_value); int hash_intptr_addintptr(struct hash_intptr * * p_hash_intptr, const void * p_strkey, unsigned int p_strlen, int p_int_value, void * p_ptr_value) { int ret; struct hash_intptr * fg_new_hash; struct hash_intptr * fg_hash, * fg_old; ret = hash_check_args((void *) p_hash_intptr, p_strkey, p_strlen, " "); if (ret < 0) return HASH_INTPTR_ERROR_ARGS; fg_new_hash = NULL; fg_hash = fg_old = *p_hash_intptr; HASH_FIND(hh, fg_hash, p_strkey, p_strlen, fg_new_hash); if (fg_new_hash == NULL) { fg_new_hash = (struct hash_intptr *) calloc(0x1, sizeof(*fg_new_hash)); if (fg_new_hash == NULL) { fputs("Error, system out of memory!\n", stderr); fflush(stderr); if (fg_hash != fg_old) *p_hash_intptr = fg_hash; return HASH_INTPTR_ERROR_MEMORY; } fg_new_hash->hh_keylen = p_strlen; fg_new_hash->hh_int = p_int_value; fg_new_hash->hh_ptr = p_ptr_value; memcpy(fg_new_hash->hh_key, p_strkey, p_strlen); if (p_strlen < HASH_INTPTR_MAXLEN) fg_new_hash->hh_key[p_strlen] = (unsigned char) 0; HASH_ADD(hh, fg_hash, hh_key, p_strlen, fg_new_hash); } else { if (fg_new_hash->hh_keylen != p_strlen) { fprintf(stderr, "FATAL ERROR, keys: %p, %p, keylen: %u, %u\n", fg_new_hash->hh_key, p_strkey, fg_new_hash->hh_keylen, p_strlen); fflush(stderr); } fg_new_hash->hh_int = p_int_value; fg_new_hash->hh_ptr = p_ptr_value; } if (fg_hash != fg_old) *p_hash_intptr = fg_hash; return HASH_INTPTR_ERROR_OK; } int hash_intptr_addint(struct hash_intptr * * a_hash_intptr, const void * a_strkey, unsigned int a_strlen, int a_int_value) { return hash_intptr_addintptr(a_hash_intptr, a_strkey, a_strlen, a_int_value, NULL); } int hash_intptr_addptr(struct hash_intptr * * p_hash_intptr, const void * p_strkey, unsigned int p_strlen, void * p_ptr_value) { return hash_intptr_addintptr(p_hash_intptr, p_strkey, p_strlen, 0, p_ptr_value); } int hash_intptr_findint(struct hash_intptr * f_hash_intptr, const void * f_strkey, unsigned int f_strlen, int * f_int_value) { int ret; struct hash_intptr * fg_new_hash; if (f_hash_intptr == NULL) return HASH_INTPTR_ERROR_NOTFOUND; ret = hash_check_args((void *) f_hash_intptr, f_strkey, f_strlen, " "); if (ret < 0) return HASH_INTPTR_ERROR_ARGS; fg_new_hash = NULL; HASH_FIND(hh, f_hash_intptr, f_strkey, f_strlen, fg_new_hash); if (fg_new_hash == NULL) return HASH_INTPTR_ERROR_NOTFOUND; if (f_int_value != NULL) *f_int_value = fg_new_hash->hh_int; return HASH_INTPTR_ERROR_OK; } int hash_intptr_findptr(struct hash_intptr * f_hash_intptr, const void * f_strkey, unsigned int f_strlen, void * * f_ptr_value) { int ret; struct hash_intptr * fg_new_hash; if (f_hash_intptr == NULL) return HASH_INTPTR_ERROR_NOTFOUND; ret = hash_check_args((void *) f_hash_intptr, f_strkey, f_strlen, " "); if (ret < 0) return HASH_INTPTR_ERROR_ARGS; fg_new_hash = NULL; HASH_FIND(hh, f_hash_intptr, f_strkey, f_strlen, fg_new_hash); if (fg_new_hash == NULL) return HASH_INTPTR_ERROR_NOTFOUND; if (f_ptr_value != NULL) *f_ptr_value = fg_new_hash->hh_ptr; return HASH_INTPTR_ERROR_OK; } int hash_intptr_remove(struct hash_intptr * * r_hash_intptr, const void * r_strkey, unsigned int r_strlen) { int ret; struct hash_intptr * fg_old_hash; struct hash_intptr * fg_hash, * fg_old; ret = hash_check_args((void *) r_hash_intptr, r_strkey, r_strlen, " "); if (ret < 0) return HASH_INTPTR_ERROR_ARGS; fg_old_hash = NULL; fg_hash = fg_old = *r_hash_intptr; if (fg_hash == NULL) return HASH_INTPTR_ERROR_OK; HASH_FIND(hh, fg_hash, r_strkey, r_strlen, fg_old_hash); if (fg_old_hash == NULL) return HASH_INTPTR_ERROR_OK; HASH_DEL(fg_hash, fg_old_hash); fg_old_hash->hh_key[0] = (unsigned char) 0; fg_old_hash->hh_keylen = 0; fg_old_hash->hh_int = 0; fg_old_hash->hh_ptr = NULL; free(fg_old_hash); fg_old_hash = NULL; if (fg_hash != fg_old) *r_hash_intptr = fg_hash; return HASH_INTPTR_ERROR_OK; } int hash_intptr_count(struct hash_intptr * c_hash_intptr) { int rval = 0; if (c_hash_intptr != NULL) { rval = HASH_COUNT(c_hash_intptr); } return rval; } int hash_intptr_iter(struct hash_intptr * * i_hash_intptr, int (* iter_func)(struct hash_intptr *, int, void *), void * whatp) { int rval; struct hash_intptr * fg_iter, * fg_tmp; struct hash_intptr * fg_hash, * fg_old; rval = 0; if (i_hash_intptr == NULL) return HASH_INTPTR_ERROR_ARGS; if (iter_func == NULL) return HASH_INTPTR_ERROR_ARGS; fg_iter = fg_tmp = NULL; fg_hash = fg_old = *i_hash_intptr; if (fg_hash == NULL) return rval; HASH_ITER(hh, fg_hash, fg_iter, fg_tmp) { if (iter_func(fg_iter, rval, whatp) < 0) break; rval++; } if (fg_old != fg_hash) *i_hash_intptr = fg_hash; return rval; } void hash_intptr_removeall(struct hash_intptr * * ra_hash_intptr) { struct hash_intptr * fg_iter, * fg_tmp; struct hash_intptr * fg_hash, * fg_old; if (ra_hash_intptr == NULL) return; fg_iter = fg_tmp = NULL; fg_hash = fg_old = *ra_hash_intptr; if (fg_hash == NULL) return; HASH_ITER(hh, fg_hash, fg_iter, fg_tmp) { HASH_DEL(fg_hash, fg_iter); fg_iter->hh_key[0] = (unsigned char) 0; fg_iter->hh_keylen = 0; fg_iter->hh_int = 0; fg_iter->hh_ptr = NULL; free(fg_iter); } if (fg_hash != fg_old) *ra_hash_intptr = fg_hash; } typedef struct hash_intptr hash_intptr_t; static hash_intptr_t *insert_stmt_hash; static void str_replace_chr(char *line) { for(int i = 0; line[i] != '\0';i++) { if(!( (line[i] >= '0' && line[i] <= '9') || (line[i] >= 'a' && line[i] <= 'z') || (line[i] >= 'A' && line[i] <= 'Z') || line[i] == '\0'|| line[i] == '_')) { line[i] = '_'; } } } static size_t getEnd_backend_sqlite(UA_Server *server, void *context, const UA_NodeId *sessionId, void *sessionContext, const UA_NodeId * nodeId); static size_t resultSize_backend_sqlite(UA_Server *server, void *context, const UA_NodeId *sessionId, void *sessionContext, const UA_NodeId * nodeId, size_t startIndex, size_t endIndex) { sqlite3 *db = (sqlite3*)context; if(db == NULL) { printf("resultSize_backend_sqlite 0000000000000000000\n"); return 0; } int nrow=0; int ncolumn = 0; char *zErrMsg =NULL; char **azResult=NULL; //二维数组存放结果 char sql[512] = {0}; char buff[128] = {0}; memcpy(buff,nodeId->identifier.string.data,nodeId->identifier.string.length); sprintf(sql,"select count(*) from 'variable' where name = '%s'",buff); sqlite3_get_table( db , sql , &azResult , &nrow , &ncolumn , &zErrMsg ); if(zErrMsg ) { printf("resultSize_backend_sqlite error:%s\n",zErrMsg); sqlite3_free_table(azResult); sqlite3_free(zErrMsg); return 0; } int rowcount = 0; if(nrow == 1) { rowcount = (int)atoi(azResult[1]); } //printf("resultSize_backend_memory rowcount:%d startindex:%d endindex:%d\n",rowcount,(int)startIndex,(int)endIndex); sqlite3_free_table(azResult); sqlite3_free(zErrMsg); if (rowcount == 0 || startIndex == (size_t)rowcount || endIndex == (size_t)rowcount) return 0; return endIndex - startIndex + 1; } static size_t getDateTimeMatch_backend_sqlite(UA_Server *server, void *context, const UA_NodeId *sessionId, void *sessionContext, const UA_NodeId * nodeId, const UA_DateTime timestamp, const MatchStrategy strategy) { //printf("getDateTimeMatch_backend_sqlite 11111111111111\n"); sqlite3 *db = (sqlite3*)context; if(db == NULL) { printf("getDateTimeMatch_backend_sqlite 0000000000000000000\n"); return 0; } int nrow=0; int ncolumn = 0; char *zErrMsg =NULL; char **azResult=NULL; //二维数组存放结果 char sql[512] = {0}; char buff[128] = {0}; memcpy(buff,nodeId->identifier.string.data,nodeId->identifier.string.length); sprintf(sql,"select count(*) from 'variable' where name = '%s' and time <= %" PRId64"",buff,(uint64_t)timestamp); sqlite3_get_table( db , sql , &azResult , &nrow , &ncolumn , &zErrMsg ); if(zErrMsg ) { printf("getDateTimeMatch_backend_sqlite error:%s\n",zErrMsg); sqlite3_free_table(azResult); sqlite3_free(zErrMsg); return 0; } int rowcount = 0; size_t storeEnd = getEnd_backend_sqlite(server,context,sessionId,sessionContext,nodeId) -1; if(nrow == 1) { rowcount = (int)atoi(azResult[1]); } sqlite3_free_table(azResult); sqlite3_free(zErrMsg); if ((strategy == MATCH_EQUAL\ || strategy == MATCH_EQUAL_OR_AFTER\ || strategy == MATCH_EQUAL_OR_BEFORE)) if(rowcount == 0) return 0; return (size_t)rowcount-1; switch (strategy) { case MATCH_AFTER: return (size_t)rowcount; case MATCH_EQUAL_OR_AFTER: if(rowcount == 0) return 0; return (size_t)rowcount-1; case MATCH_EQUAL_OR_BEFORE: // retval == true aka "equal" is handled before // Fall through if !retval case MATCH_BEFORE: if (rowcount > 1){ return (size_t)rowcount-2; } else { return storeEnd; } default: break; } return storeEnd; } static char* GetDataValueToStr(const UA_Variant *data) { char *result = (char*)malloc(128); if (result == NULL) return NULL; memset(result,0,128); if(data->type == &UA_TYPES[UA_TYPES_BOOLEAN]){ UA_Boolean temp1 = *(UA_Boolean*)data->data; if(temp1 == UA_TRUE) sprintf(result,"%d",1); else sprintf(result,"%d",0); } else if(data->type == &UA_TYPES[UA_TYPES_INT16]) { int16_t temp1 = *(UA_Int16*)data->data; sprintf(result,"%d",temp1); } else if(data->type == &UA_TYPES[UA_TYPES_INT32]) { int32_t temp1 = *(UA_Int32*)data->data; sprintf(result,"%d",temp1); } else if(data->type == &UA_TYPES[UA_TYPES_INT64]) { int64_t temp1 = *(UA_Int64*)data->data; sprintf(result,"%lld",temp1); } else if(data->type == &UA_TYPES[UA_TYPES_FLOAT]) { float temp1 = *(UA_Float*)data->data; //printf("curwatchdata->format:%s\n",curwatchdata->format); sprintf(result,"%f",temp1); } else if(data->type == &UA_TYPES[UA_TYPES_DOUBLE]) { double temp1 = *(UA_Double*)data->data; sprintf(result,"%f",temp1); } else if(data->type == &UA_TYPES[UA_TYPES_UINT16]) { uint16_t temp1 = *(UA_UInt16*)data->data; sprintf(result,"%u",temp1); } else if(data->type == &UA_TYPES[UA_TYPES_UINT32]) { uint32_t temp1 = *(UA_UInt32*)data->data; sprintf(result,"%u",temp1); } else if(data->type == &UA_TYPES[UA_TYPES_UINT64]) { uint64_t temp1 = *(UA_UInt64*)data->data; sprintf(result,"%" PRId64"",temp1); } else if(data->type == &UA_TYPES[UA_TYPES_STRING]) { UA_String *name = (UA_String*)data->data; sprintf(result,"%s",name->data); }else { free(result); return NULL; } return result; } static int GetDataTypeToInt(const UA_Variant *data) { if(data->type == &UA_TYPES[UA_TYPES_BOOLEAN]) { return UA_TYPES_BOOLEAN; } else if(data->type == &UA_TYPES[UA_TYPES_INT16]) { return UA_TYPES_INT16; } else if(data->type == &UA_TYPES[UA_TYPES_INT32]) { return UA_TYPES_INT32; } else if(data->type == &UA_TYPES[UA_TYPES_INT64]) { return UA_TYPES_INT64; } else if(data->type == &UA_TYPES[UA_TYPES_FLOAT]) { return UA_TYPES_FLOAT; } else if(data->type == &UA_TYPES[UA_TYPES_DOUBLE]) { return UA_TYPES_DOUBLE; } else if(data->type == &UA_TYPES[UA_TYPES_UINT16]) { return UA_TYPES_UINT16; } else if(data->type == &UA_TYPES[UA_TYPES_UINT32]) { return UA_TYPES_UINT32; } else if(data->type == &UA_TYPES[UA_TYPES_UINT64]) { return UA_TYPES_UINT64; } else if(data->type == &UA_TYPES[UA_TYPES_STRING]) { return UA_TYPES_STRING; }else { return UA_TYPES_INT16; } } #if 1 // 写任务的实现 #define MAX_WRITE_LIST_LEN (30000) #define NODE_ID_MAX_LEN (128) typedef struct list_it { UA_DateTime timestamp; char nodeid_to_str[NODE_ID_MAX_LEN]; unsigned int nodeid_to_str_len; int ua_type; char *ua_value_to_str; // struct list_it *next; } list_it; typedef struct write_list { pthread_rwlock_t rwlock; list_it *head; list_it *tail; sem_t write_sem; int cur_len; int is_ok; } write_list; static write_list *local_list; #define get_write_list_len(p_list) (p_list->cur_len) static write_list *new_write_list(void) { write_list *this_p = (write_list *)calloc(1, sizeof(write_list)); if (this_p != NULL) { pthread_rwlock_init(&this_p->rwlock, NULL); sem_init(&this_p->write_sem, 0, 0); this_p->head = NULL; this_p->tail = NULL; this_p->is_ok = 1; this_p->cur_len = 0; } return this_p; } static int write_list_deinit(write_list *this_p) { sem_destroy(&this_p->write_sem); // pthread_rwlock_wrlock(&this_p->rwlock); while (this_p->head != NULL) { free(this_p->head); this_p->head = this_p->head->next; } pthread_rwlock_unlock(&this_p->rwlock); pthread_rwlock_destroy(&this_p->rwlock); this_p->is_ok = 0; this_p->cur_len = 0; return 0; } static int write_list_push(write_list *this_p, list_it *it) { pthread_rwlock_wrlock(&this_p->rwlock); if (this_p->head == NULL) { this_p->head = it; this_p->tail = it; this_p->cur_len = 1; } else { this_p->tail->next = it; this_p->tail = it; this_p->cur_len++; } pthread_rwlock_unlock(&this_p->rwlock); return sem_post(&this_p->write_sem); } static list_it *write_list_pop(write_list *this_p, uint32_t timeout_ms) { struct timespec now; struct timespec wt; list_it *it = NULL; clock_gettime(CLOCK_REALTIME, &now); wt.tv_sec = now.tv_sec + timeout_ms / 1000; wt.tv_nsec = now.tv_nsec + (timeout_ms % 1000) * 1000 * 1000; if (wt.tv_nsec >= 1000000000) { wt.tv_nsec -= 1000000000; wt.tv_sec += 1; } if (0 == sem_timedwait(&this_p->write_sem, &wt)) { pthread_rwlock_wrlock(&this_p->rwlock); if (this_p->head != NULL) { it = this_p->head; this_p->head = it->next; this_p->cur_len--; } pthread_rwlock_unlock(&this_p->rwlock); } return it; } //正常返回0，list超限制长度了返回-9999，其他异常返回其他负数 static int save_to_write_list(UA_DateTime *ua_t, const UA_NodeId *nodeId, const UA_DataValue *value, void *out_cur_mem) { struct opcua_write_limit // 这个结构不能变，在opcua库中也是按这个结构输出的 { const int limit; int cur; } *p_limit = (struct opcua_write_limit *)out_cur_mem; int ret = -1; unsigned int length = (unsigned int)nodeId->identifier.string.length; if (length >= NODE_ID_MAX_LEN) { return -2; } if (local_list != NULL) { int limit = (p_limit == NULL) ? MAX_WRITE_LIST_LEN : p_limit->limit; if (get_write_list_len(local_list) < limit) { char *tmp = GetDataValueToStr(&value->value); if (tmp == NULL) { if (tmp[0] == '\0') { free(tmp); } return -3; } // list_it *it = (list_it *)calloc(1, sizeof(list_it)); it->next = NULL; it->ua_value_to_str = tmp; it->timestamp = *ua_t; it->ua_type = GetDataTypeToInt(&value->value); memcpy(it->nodeid_to_str, nodeId->identifier.string.data, length); it->nodeid_to_str[length] = 0; it->nodeid_to_str_len = length; ret = write_list_push(local_list, it); if (p_limit != NULL) { p_limit->cur = get_write_list_len(local_list); } } else { ret = -9999; } } else { ret = -4; } return ret; } void *open62541_write_thread(void *args) { sqlite3 *db = (sqlite3 *)args; list_it *it = NULL; char insertSQL[512] = {0}; sqlite3_stmt *insert_stmt = NULL; char replaced_str[NODE_ID_MAX_LEN] = {0}; while (local_list->is_ok != 0) { it = write_list_pop(local_list, 2000); if (it != NULL) { if (hash_intptr_findptr(insert_stmt_hash, it->nodeid_to_str, it->nodeid_to_str_len, (void **)&insert_stmt) < 0) { memcpy(replaced_str, it->nodeid_to_str, it->nodeid_to_str_len); replaced_str[it->nodeid_to_str_len] = 0; str_replace_chr(replaced_str); sprintf(insertSQL, "INSERT INTO '%s' VALUES(NULL,?,?,?)", replaced_str); int ret = sqlite3_prepare_v2(db, insertSQL, (int)strlen(insertSQL), &insert_stmt, NULL); if (ret != SQLITE_OK) { printf("sqlite3_prepare_v2 error:%d, insertSQL:%s\n", ret, insertSQL); if (insert_stmt) { sqlite3_finalize(insert_stmt); } } else { ret = hash_intptr_addptr(&insert_stmt_hash, it->nodeid_to_str, it->nodeid_to_str_len, insert_stmt); if (ret < 0) { printf("hash_intptr_addptr failed, key: %s\n", it->nodeid_to_str); } } } sqlite3_bind_int(insert_stmt, 1, it->ua_type); sqlite3_bind_text(insert_stmt, 2, it->ua_value_to_str, -1, NULL); sqlite3_bind_int64(insert_stmt, 3, it->timestamp); if (sqlite3_step(insert_stmt) != SQLITE_DONE) { printf("sqlite3_step error\n"); } else { sqlite3_reset(insert_stmt); } // free(it->ua_value_to_str); free(it); } else { sleep(1); } } write_list_deinit(local_list); local_list = NULL; return NULL; } #endif static UA_StatusCode serverSetHistoryData_backend_sqlite(UA_Server *server, void *context, const UA_NodeId *sessionId, void *sessionContext, const UA_NodeId * nodeId, UA_Boolean historizing, const UA_DataValue *value) { int locked; sqlite3 *db = (sqlite3*)context; if(db == NULL) { printf("serverSetHistoryData_backend_sqlite 0000000000000000000\n"); return UA_STATUSCODE_GOOD; } UA_DateTime timestamp = 0; if (value->hasSourceTimestamp) { timestamp = value->sourceTimestamp; } else if (value->hasServerTimestamp) { timestamp = value->serverTimestamp; } else { timestamp = UA_DateTime_now(); } if(nodeId->identifierType != UA_NODEIDTYPE_STRING) { return UA_STATUSCODE_GOOD; } // locked = save_to_write_list(×tamp, nodeId, value, sessionContext); if (locked == 0) // 没有异常 { return UA_STATUSCODE_GOOD; } else if (locked == -9999) // 队列超过限制了 { return UA_STATUSCODE_GOOD; } else { // 写队列是没有异常或写队列满了就直接返回，否则使用下面的接口 } return UA_STATUSCODE_GOOD; } static size_t getEnd_backend_sqlite(UA_Server *server, void *context, const UA_NodeId *sessionId, void *sessionContext, const UA_NodeId * nodeId) { sqlite3 *db = (sqlite3*)context; if(db == NULL) { printf("getEnd_backend_sqlite 0000000000000000000\n"); return UA_STATUSCODE_GOOD; } size_t buflen; int rowcount = 0; int nrow=0; int ncolumn = 0; char *zErrMsg =NULL; char **azResult=NULL; //二维数组存放结果 char sql[512] = {0}; char buff[128] = {0}; if (nodeId->identifierType == UA_NODEIDTYPE_NUMERIC) return rowcount; buflen = (size_t) nodeId->identifier.string.length; if (buflen >= sizeof(buff)) buflen = (size_t) (sizeof(buff) - 1); if (buflen > 0) memcpy(buff, nodeId->identifier.string.data, buflen); snprintf(sql, sizeof(sql) - 1, "select count(*) from 'variable' where name = '%s'", buff); sqlite3_get_table( db , sql , &azResult , &nrow , &ncolumn , &zErrMsg ); if(zErrMsg ) { printf("getEnd_backend_sqlite error:%s\n",zErrMsg); sqlite3_free_table(azResult); sqlite3_free(zErrMsg); return UA_STATUSCODE_GOOD; } if(nrow == 1) { rowcount = (int)atoi(azResult[1]); } sqlite3_free_table(azResult); sqlite3_free(zErrMsg); return (size_t)rowcount; } static size_t lastIndex_backend_sqlite(UA_Server *server, void *context, const UA_NodeId *sessionId, void *sessionContext, const UA_NodeId * nodeId) { sqlite3 *db = (sqlite3*)context; if(db == NULL) { printf("lastIndex_backend_sqlite 0000000000000000000\n"); return UA_STATUSCODE_GOOD; } int rowcount = 0; int nrow=0; int ncolumn = 0; char *zErrMsg =NULL; char **azResult=NULL; //二维数组存放结果 char sql[512] = {0}; char buff[128] = {0}; size_t buflen; if (nodeId->identifierType == UA_NODEIDTYPE_NUMERIC) return rowcount; buflen = (size_t) nodeId->identifier.string.length; if (buflen >= sizeof(buff)) buflen = (size_t) (sizeof(buff) - 1); if (buflen > 0) memcpy(buff, nodeId->identifier.string.data, buflen); snprintf(sql, sizeof(sql) - 1, "select count(*) from 'variable' where name = '%s'", buff); sqlite3_get_table( db , sql , &azResult , &nrow , &ncolumn , &zErrMsg ); if(zErrMsg ) { printf("lastIndex_backend_sqlite error:%s\n",zErrMsg); sqlite3_free_table(azResult); sqlite3_free(zErrMsg); return 0; } if(nrow == 1) { rowcount = (int)atoi(azResult[1]); } //printf("rowcount:%d\n",rowcount); sqlite3_free_table(azResult); sqlite3_free(zErrMsg); return (size_t)rowcount-1; } static size_t firstIndex_backend_sqlite(UA_Server *server, void *context, const UA_NodeId *sessionId, void *sessionContext, const UA_NodeId * nodeId) { return 0; } static UA_Boolean boundSupported_backend_sqlite(UA_Server *server, void *context, const UA_NodeId *sessionId, void *sessionContext, const UA_NodeId * nodeId) { return true; } static UA_Boolean timestampsToReturnSupported_backend_sqlite(UA_Server *server, void *context, const UA_NodeId *sessionId, void *sessionContext, const UA_NodeId *nodeId, const UA_TimestampsToReturn timestampsToReturn) { return true; } static size_t getDataValues_backend_sqlite(UA_Server *server, void *context, const UA_NodeId *sessionId, void *sessionContext, const UA_NodeId * nodeId, size_t index, size_t total,UA_Boolean reverse, UA_DataValue * result) { sqlite3 *db = (sqlite3*)context; int nrow=0; int i=0; int ncolumn = 0; char *zErrMsg =NULL; char **azResult=NULL; //二维数组存放结果 char sql[512] = {0}; char buff[128] = {0}; memcpy(buff,nodeId->identifier.string.data,nodeId->identifier.string.length); if(reverse) { sprintf(sql,"select datatype,result,time from 'variable' where name = '%s' order by time desc limit %d offset %d",buff,(int)total,(int)index); } else { sprintf(sql,"select datatype,result,time from 'variable' where name = '%s' order by time limit %d offset %d",buff,(int)total,(int)index); } printf("sql:%s\n",sql); sqlite3_get_table( db , sql , &azResult , &nrow , &ncolumn , &zErrMsg ); printf("nrow:%d ncolumn:%d\n",nrow,ncolumn); //数据类型 if(nrow == 0) return 0; printf("22222222222222\n"); for(i=0;iidentifier.string.data,nodeId->identifier.string.length); sprintf(sql,"select datatype,result,time from 'variable' where name = '%s' order by time limit 1 offset %d",buff,(int)index); sqlite3_get_table( db , sql , &azResult , &nrow , &ncolumn , &zErrMsg ); //printf("sql:%s\n",sql); //printf("nrow:%d ncolumn:%d\n",nrow,ncolumn); //数据类型 if(nrow == 0) { sqlite3_free_table(azResult); sqlite3_free(zErrMsg); return NULL; } UA_DataValue *result = (UA_DataValue*)UA_malloc(sizeof(UA_DataValue)); UA_DataValue_init(result); result->hasValue = true; UA_Variant_init(&result->value); //printf("time:%s\n",azResult[5]); UA_DateTime timestamp = (UA_DateTime)strtoull(azResult[5],NULL,10); //printf("timestamp:%" PRId64"\n",timestamp); switch(atoi(azResult[3])) { case UA_TYPES_BOOLEAN: { UA_Boolean temp1 = atoi(azResult[4]); UA_Variant_setScalar(&result->value, &temp1, &UA_TYPES[UA_TYPES_BOOLEAN]); break; } case UA_TYPES_INT16: { int16_t temp1 = (int16_t)atoi(azResult[4]); UA_Variant_setScalar(&result->value, &temp1, &UA_TYPES[UA_TYPES_INT16]); break; } case UA_TYPES_INT32: { int32_t temp1 = (int32_t)atol(azResult[4]); UA_Variant_setScalar(&result->value, &temp1, &UA_TYPES[UA_TYPES_INT32]); break; } case UA_TYPES_INT64: { int64_t temp1 = (int64_t)atol(azResult[4]); UA_Variant_setScalar(&result->value, &temp1, &UA_TYPES[UA_TYPES_INT64]); break; } case UA_TYPES_FLOAT: { float temp1 = (float)atof(azResult[4]); UA_Variant_setScalar(&result->value, &temp1, &UA_TYPES[UA_TYPES_FLOAT]); break; } case UA_TYPES_DOUBLE: { double temp1 = strtod(azResult[4],NULL); UA_Variant_setScalar(&result->value, &temp1, &UA_TYPES[UA_TYPES_DOUBLE]); break; } case UA_TYPES_UINT16: { uint16_t temp1 = (uint16_t)strtoul(azResult[4],NULL,10); UA_Variant_setScalar(&result->value, &temp1, &UA_TYPES[UA_TYPES_UINT16]); break; } case UA_TYPES_UINT32: { uint32_t temp1 = (uint32_t)strtoul(azResult[4],NULL,10); UA_Variant_setScalar(&result->value, &temp1, &UA_TYPES[UA_TYPES_UINT32]); break; } case UA_TYPES_UINT64: { uint64_t temp1 = strtoul(azResult[4],NULL,10); UA_Variant_setScalar(&result->value, &temp1, &UA_TYPES[UA_TYPES_UINT64]); break; } case UA_TYPES_STRING: { UA_String ua_result; ua_result.length = strlen(azResult[4])+1; ua_result.data = (UA_Byte *)malloc(ua_result.length+1); strcpy((char*)ua_result.data, azResult[4]); UA_Variant_setScalarCopy(&result->value, &ua_result, &UA_TYPES[UA_TYPES_STRING]); UA_String_clear(&ua_result); free(ua_result.data); break; } } sqlite3_free_table(azResult); sqlite3_free(zErrMsg); result->sourceTimestamp = timestamp; result->hasSourceTimestamp = true; result->serverTimestamp = timestamp; result->hasServerTimestamp = true; result->hasSourcePicoseconds = false; result->hasServerPicoseconds = false; result->hasStatus = true; result->status = UA_STATUSCODE_GOOD; return result; } static UA_StatusCode insertDataValue_backend_sqlite(UA_Server *server, void *hdbContext, const UA_NodeId *sessionId, void *sessionContext, const UA_NodeId *nodeId, const UA_DataValue *value) { return UA_STATUSCODE_GOOD; } static UA_StatusCode replaceDataValue_backend_sqlite(UA_Server *server, void *hdbContext, const UA_NodeId *sessionId, void *sessionContext, const UA_NodeId *nodeId, const UA_DataValue *value) { return UA_STATUSCODE_GOOD; } static UA_StatusCode updateDataValue_backend_sqlite(UA_Server *server, void *hdbContext, const UA_NodeId *sessionId, void *sessionContext, const UA_NodeId *nodeId, const UA_DataValue *value) { return UA_STATUSCODE_GOODENTRYINSERTED; } static UA_StatusCode removeDataValue_backend_sqlite(UA_Server *server, void *hdbContext, const UA_NodeId *sessionId, void *sessionContext, const UA_NodeId *nodeId, UA_DateTime startTimestamp, UA_DateTime endTimestamp) { return UA_STATUSCODE_GOOD; } static void deleteMembers_backend_sqlite(UA_HistoryDataBackend *backend) { if (backend == NULL || backend->context == NULL) return; //UA_sqliteStoreContext_deleteMembers((UA_sqliteStoreContext*)backend->context); sqlite3_close((sqlite3*)backend->context); } static UA_StatusCode getHistoryData_backend_sqlite(UA_Server *server, const UA_NodeId *sessionId, void *sessionContext, const UA_HistoryDataBackend *backend, const UA_DateTime start, const UA_DateTime end, const UA_NodeId *nodeId, size_t maxSizePerResponse, UA_UInt32 numValuesPerNode, UA_Boolean returnBounds, UA_TimestampsToReturn timestampsToReturn, UA_NumericRange range, UA_Boolean releaseContinuationPoints, const UA_ByteString *continuationPoint, UA_ByteString *outContinuationPoint, UA_HistoryData *result) { // printf("start:%" PRId64" end:%" PRId64" maxSizePerResponse:%d numValuesPerNode:%u \n",start,end,(int)maxSizePerResponse,numValuesPerNode); size_t skip = 0; if (continuationPoint->length > 0) { if (continuationPoint->length >= sizeof(size_t)) { skip = *((size_t*)(continuationPoint->data)); } else { return UA_STATUSCODE_BADCONTINUATIONPOINTINVALID; } } if(releaseContinuationPoints) { printf("release continuationPoint\n"); UA_ByteString_deleteMembers((UA_ByteString *)continuationPoint); } sqlite3 *db = (sqlite3*)backend->context; if(db == NULL) { printf("getHistoryData_backend_sqlite 0000000000000000000\n"); return UA_STATUSCODE_GOOD; } int i=0; int nrow=0; int ncolumn = 0; char *zErrMsg =NULL; char **azResult=NULL; //二维数组存放结果 char sql[512] = {0}; char buff[128] = {0}; memcpy(buff,nodeId->identifier.string.data,nodeId->identifier.string.length); str_replace_chr(buff); char sqlCount[512] = {0}; size_t maxSize = maxSizePerResponse ; // if(numValuesPerNode > 0 && maxSizePerResponse > numValuesPerNode) // { // maxSize = numValuesPerNode ; // } if(returnBounds) { if(start < end) { sprintf(sql,"select datatype,result,time from '%s' where time >= %" PRId64" and time <=%" PRId64" order by time limit %d offset %d",buff,start,end,maxSize,(int)skip); sprintf(sqlCount,"select count(*) from '%s' where time >= %" PRId64" and time <=%" PRId64"",buff,start,end); } else if(end == 0) { sprintf(sql,"select datatype,result,time from '%s' where time >= %" PRId64" order by time limit %d offset %d",buff,start,maxSize,(int)skip); sprintf(sqlCount,"select count(*) from '%s' where time >= %" PRId64"",buff,start); } else { sprintf(sql,"select datatype,result,time from '%s' where time >= %" PRId64" and time <=%" PRId64" order by time limit %d offset %d",buff,end,start,maxSize,(int)skip); sprintf(sqlCount,"select count(*) from '%s' where time >= %" PRId64" and time <=%" PRId64"",buff,end,start); } } else { if(start < end) { sprintf(sql,"select datatype,result,time from '%s' where time > %" PRId64" and time <%" PRId64" order by time limit %d offset %d",buff,start,end,maxSize,(int)skip); sprintf(sqlCount,"select count(*) from '%s' where time > %" PRId64" and time <%" PRId64"",buff,start,end); } else if(end == 0) { sprintf(sql,"select datatype,result,time from '%s' where time > %" PRId64" order by time limit %d offset %d",buff,start,maxSize,(int)skip); sprintf(sqlCount,"select count(*) from '%s' where time > %" PRId64"",buff,start); } else { sprintf(sql,"select datatype,result,time from '%s' where time > %" PRId64" and time <%" PRId64" order by time limit %d offset %d",buff,end,start,maxSize,(int)skip); sprintf(sqlCount,"select count(*) from '%s' where time > %" PRId64" and time <%" PRId64"",buff,end,start); } } sqlite3_get_table( db , sqlCount , &azResult , &nrow , &ncolumn , &zErrMsg ); if(zErrMsg ) { printf("getHistoryData_backend_sqlite 1111111111111 error:%s\n",zErrMsg); sqlite3_free_table(azResult); sqlite3_free(zErrMsg); return UA_STATUSCODE_BADOUTOFMEMORY; } int totalCount = (int)atoi(azResult[1]); // printf("totalCount:%d %s\n",totalCount,sqlCount); sqlite3_free_table(azResult); sqlite3_free(zErrMsg); //printf("sql:%s\n",sql); sqlite3_get_table( db , sql , &azResult , &nrow , &ncolumn , &zErrMsg ); //printf("return rows:%d\n",nrow); if(nrow == 0 ) { //printf("sqlite3_get_table error:%s\n",sql); sqlite3_free_table(azResult); sqlite3_free(zErrMsg); result->dataValuesSize = nrow; result->dataValues = NULL; return UA_STATUSCODE_GOOD; } UA_DataValue *outResult= (UA_DataValue*)UA_Array_new(nrow, &UA_TYPES[UA_TYPES_DATAVALUE]); if (!outResult || zErrMsg ) { printf("getHistoryData_backend_sqlite 2222222222222222 error:%s\n",zErrMsg); sqlite3_free_table(azResult); sqlite3_free(zErrMsg); return UA_STATUSCODE_BADOUTOFMEMORY; } //printf("22222222222222\n"); for(i=0;idataValuesSize = nrow; result->dataValues = outResult; // there are more values if ((int)skip + nrow < totalCount && (int)skip + nrow < (int)maxSize) { if(UA_ByteString_allocBuffer(outContinuationPoint, sizeof(size_t) + sizeof(size_t)) != UA_STATUSCODE_GOOD) { return UA_STATUSCODE_BADOUTOFMEMORY; } outContinuationPoint->length = sizeof(size_t); size_t t = sizeof(size_t); outContinuationPoint->data = (UA_Byte*)UA_malloc(t); *((size_t*)(outContinuationPoint->data)) = skip + nrow; } return UA_STATUSCODE_GOOD; } void creat_table(sqlite3 *db,char *node_name) //这个在外部实现的有可以不要 { /* 创建表 */ char *zErrMsg =NULL; char sql[256] = {0}; str_replace_chr(node_name); sprintf(sql," CREATE TABLE %s (\ ID INTEDER PRIMARY KEY,\ datatype INTEGER,\ result VARCHAR(64),\ time INTEGER\ );",node_name); sqlite3_exec(db,sql,NULL,NULL,&zErrMsg); sqlite3_free(zErrMsg); /* 创建索引 */ sprintf(sql,"CREATE INDEX variable_index ON %s (time);",node_name); sqlite3_exec(db,sql,NULL,NULL,&zErrMsg); sqlite3_free(zErrMsg); } UA_HistoryDataBackend UA_HistoryDataBackend_Sqlite(char* path) { //初始化数据库 sqlite3 *db; if (sqlite3_open(path, &db)) { printf("Can't open database: %s\n", sqlite3_errmsg(db)); exit(0); } char *zErrMsg =NULL; /* 优化性能 */ sqlite3_exec(db,"PRAGMA main.cache_size=100;",NULL,NULL,&zErrMsg); sqlite3_free(zErrMsg); sqlite3_exec(db,"PRAGMA temp.cache_size=100;",NULL,NULL,&zErrMsg); sqlite3_free(zErrMsg); sqlite3_exec(db,"PRAGMA cache_size=1000;",NULL,NULL,&zErrMsg); sqlite3_free(zErrMsg); sqlite3_exec(db,"PRAGMA journal_mode = WAL;",NULL,NULL,&zErrMsg); sqlite3_free(zErrMsg); sqlite3_exec(db,"PRAGMA journal_size_limit = 16384;",NULL,NULL,&zErrMsg); sqlite3_free(zErrMsg); sqlite3_exec(db,"PRAGMA busy_timeout=500;",NULL,NULL,&zErrMsg); sqlite3_free(zErrMsg); sqlite3_exec(db,"PRAGMA legacy_file_format=OFF;",NULL,NULL,&zErrMsg); sqlite3_free(zErrMsg); sqlite3_exec(db,"PRAGMA synchronous=NORMAL;",NULL,NULL,&zErrMsg); sqlite3_free(zErrMsg); sqlite3_exec(db,"PRAGMA temp_store=FILE;",NULL,NULL,&zErrMsg); sqlite3_free(zErrMsg); UA_HistoryDataBackend result; memset(&result, 0, sizeof(UA_HistoryDataBackend)); result.serverSetHistoryData = &serverSetHistoryData_backend_sqlite; result.resultSize = &resultSize_backend_sqlite; result.getEnd = &getEnd_backend_sqlite; result.lastIndex = &lastIndex_backend_sqlite; result.firstIndex = &firstIndex_backend_sqlite; result.getDateTimeMatch = &getDateTimeMatch_backend_sqlite; result.copyDataValues = NULL; result.getDataValue = &getDataValue_backend_sqlite; result.boundSupported = &boundSupported_backend_sqlite; result.timestampsToReturnSupported = ×tampsToReturnSupported_backend_sqlite; result.insertDataValue = &insertDataValue_backend_sqlite; result.updateDataValue = &updateDataValue_backend_sqlite; result.replaceDataValue = &replaceDataValue_backend_sqlite; result.removeDataValue = &removeDataValue_backend_sqlite; result.deleteMembers = &deleteMembers_backend_sqlite; result.getHistoryData = &getHistoryData_backend_sqlite; result.context = db; // printf("new_write_list......\n"); local_list = new_write_list(); pthread_t hd = 0; if (local_list == NULL) { printf("open62541 new write list err\n"); exit(1); } int ret = pthread_create(&hd, NULL, open62541_write_thread, db); if (ret != 0) { printf("failed to created pthread:open62541 write task, error: %s", strerror(errno)); exit(2); } else { pthread_setname_np(hd, "open62541_w"); } // return result; }