Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
提交 3f64b14a 编写于 作者: S Steven Li
浏览文件
操作

Merge remote-tracking branch 'origin/develop' into feature/crash_gen

上级 a6e2ef7b 29cc962f
展开全部 显示空白变更内容
内联 并排
Showing with 11302 addition and 2399 deletion
.gitignore
浏览文件 @ 3f64b14a
......@@ -46,7 +46,6 @@ html/
/CMakeCache.txt
/Makefile
/*.cmake
/deps
/src/cq/test/CMakeFiles/cqtest.dir/*.cmake
*.cmake
/src/cq/test/CMakeFiles/cqtest.dir/*.make
......@@ -65,3 +64,4 @@ CMakeError.log
/out/isenseconfig/WSL-Clang-Debug
/out/isenseconfig/WSL-GCC-Debug
/test/cfg
/src/.vs
.travis.yml
浏览文件 @ 3f64b14a
......@@ -51,7 +51,7 @@ matrix:
./test-all.sh $TRAVIS_EVENT_TYPE || travis_terminate $?
cd ${TRAVIS_BUILD_DIR}/tests/pytest
./valgrind-test.sh 2>&1 > mem-error-out.txt
./valgrind-test.sh 2>&1 > mem-error-out.log
sleep 1
# Color setting
......@@ -61,9 +61,9 @@ matrix:
GREEN_UNDERLINE='\033[4;32m'
NC='\033[0m'
grep 'start to execute\|ERROR SUMMARY' mem-error-out.txt|grep -v 'grep'|uniq|tee uniq-mem-error-out.txt
grep 'start to execute\|ERROR SUMMARY' mem-error-out.log|grep -v 'grep'|uniq|tee uniq-mem-error-out.log
for memError in `grep 'ERROR SUMMARY' uniq-mem-error-out.txt | awk '{print $4}'`
for memError in `grep 'ERROR SUMMARY' uniq-mem-error-out.log | awk '{print $4}'`
do
if [ -n "$memError" ]; then
if [ "$memError" -gt 12 ]; then
......@@ -74,8 +74,8 @@ matrix:
fi
done
grep 'start to execute\|definitely lost:' mem-error-out.txt|grep -v 'grep'|uniq|tee uniq-definitely-lost-out.txt
for defiMemError in `grep 'definitely lost:' uniq-definitely-lost-out.txt | awk '{print $7}'`
grep 'start to execute\|definitely lost:' mem-error-out.log|grep -v 'grep'|uniq|tee uniq-definitely-lost-out.log
for defiMemError in `grep 'definitely lost:' uniq-definitely-lost-out.log | awk '{print $7}'`
do
if [ -n "$defiMemError" ]; then
if [ "$defiMemError" -gt 13 ]; then
......
CMakeSettings.json
浏览文件 @ 3f64b14a
......@@ -5,7 +5,7 @@
"generator": "Unix Makefiles",
"configurationType": "Debug",
"buildRoot": "${projectDir}\\build\\",
"installRoot": "${projectDir}\\out\\install\\${name}",
"installRoot": "${projectDir}\\build\\",
"cmakeExecutable": "/usr/bin/cmake",
"cmakeCommandArgs": "",
"buildCommandArgs": "",
......
deps/CMakeLists.txt
浏览文件 @ 3f64b14a
......@@ -7,3 +7,4 @@ ADD_SUBDIRECTORY(regex)
ADD_SUBDIRECTORY(iconv)
ADD_SUBDIRECTORY(lz4)
ADD_SUBDIRECTORY(cJson)
ADD_SUBDIRECTORY(MQTT-C)
deps/MQTT-C/CMakeLists.txt 0 → 100644
浏览文件 @ 3f64b14a
cmake_minimum_required(VERSION 3.5)
project(MQTT-C VERSION 1.1.2 LANGUAGES C)
# MQTT-C build options
option(MQTT_C_OpenSSL_SUPPORT "Build MQTT-C with OpenSSL support?" OFF)
option(MQTT_C_MbedTLS_SUPPORT "Build MQTT-C with mbed TLS support?" OFF)
option(MQTT_C_EXAMPLES "Build MQTT-C examples?" ON)
option(MQTT_C_TESTS "Build MQTT-C tests?" OFF)
list (APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/cmake)
# MQTT-C library
add_library(mqttc STATIC
src/mqtt_pal.c
src/mqtt.c
)
target_include_directories(mqttc PUBLIC include)
target_link_libraries(mqttc PUBLIC
$<$<C_COMPILER_ID:MSVS>:ws2_32>
)
# Configure with OpenSSL support
if(MQTT_C_OpenSSL_SUPPORT)
find_package(OpenSSL REQUIRED)
target_link_libraries(mqttc INTERFACE OpenSSL::SSL)
target_compile_definitions(mqttc PUBLIC MQTT_USE_BIO)
endif()
# Configure with mbed TLS support
if(MQTT_C_MbedTLS_SUPPORT)
find_package(MbedTLS REQUIRED)
target_include_directories(mqttc PUBLIC ${MBEDTLS_INCLUDE_DIRS})
target_link_libraries(mqttc INTERFACE ${MBEDTLS_LIBRARY})
target_compile_definitions(mqttc PUBLIC MQTT_USE_MBEDTLS)
endif()
# Build examples
if(MQTT_C_EXAMPLES)
find_package(Threads REQUIRED)
if(MQTT_C_OpenSSL_SUPPORT)
add_executable(bio_publisher examples/bio_publisher.c)
target_link_libraries(bio_publisher Threads::Threads mqttc)
add_executable(openssl_publisher examples/openssl_publisher.c)
target_link_libraries(openssl_publisher Threads::Threads mqttc)
elseif(MQTT_C_MbedTLS_SUPPORT)
add_executable(mbedtls_publisher examples/mbedtls_publisher.c)
target_link_libraries(mbedtls_publisher Threads::Threads mqttc ${MBEDX509_LIBRARY} ${MBEDCRYPTO_LIBRARY})
else()
add_executable(simple_publisher examples/simple_publisher.c)
target_link_libraries(simple_publisher Threads::Threads mqttc)
add_executable(simple_subscriber examples/simple_subscriber.c)
target_link_libraries(simple_subscriber Threads::Threads mqttc)
add_executable(reconnect_subscriber examples/reconnect_subscriber.c)
target_link_libraries(reconnect_subscriber Threads::Threads mqttc)
endif()
endif()
# Build tests
if(MQTT_C_TESTS)
find_path(CMOCKA_INCLUDE_DIR cmocka.h)
find_library(CMOCKA_LIBRARY cmocka)
if((NOT CMOCKA_INCLUDE_DIR) OR (NOT CMOCKA_LIBRARY))
message(FATAL_ERROR "Failed to find cmocka! Add cmocka's install prefix to CMAKE_PREFIX_PATH to resolve this error.")
endif()
add_executable(tests tests.c)
target_link_libraries(tests ${CMOCKA_LIBRARY} mqttc)
target_include_directories(tests PRIVATE ${CMOCKA_INCLUDE_DIR})
endif()
# Install includes and library
# install(TARGETS mqttc
# DESTINATION lib
# )
# install(DIRECTORY include/
# DESTINATION include)
deps/MQTT-C/Doxyfile 0 → 100644
浏览文件 @ 3f64b14a
此差异已折叠。
deps/MQTT-C/LICENSE 0 → 100644
浏览文件 @ 3f64b14a
MIT License
Copyright (c) 2018 Liam Bindle
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
deps/MQTT-C/README.md 0 → 100644
浏览文件 @ 3f64b14a
<p align="right">
<a href="https://github.com/LiamBindle/MQTT-C/stargazers"><img src="https://img.shields.io/github/stars/LiamBindle/MQTT-C.svg?style=social&label=Star" style="margin-left:5em"></a>
<a href="https://github.com/LiamBindle/MQTT-C/network/members"><img src="https://img.shields.io/github/forks/LiamBindle/MQTT-C.svg?style=social&label=Fork"></a>
</p>
<p align="center">
<img width="70%" src="docs/mqtt-c-logo.png"><br>
<a href="https://liambindle.ca/MQTT-C"><img src="https://img.shields.io/badge/docs-passing-brightgreen.svg"></a>
<a href="https://github.com/LiamBindle/MQTT-C/issues"><img src="https://img.shields.io/badge/Maintained%3F-yes-green.svg"></a>
<a href="https://GitHub.com/LiamBindle/MQTT-C/issues/"><img src="https://img.shields.io/github/issues/LiamBindle/MQTT-C.svg"></a>
<a href="https://github.com/LiamBindle/MQTT-C/issues"><img src="https://img.shields.io/github/issues-closed/LiamBindle/MQTT-C.svg"></a>
<a href="https://github.com/LiamBindle/MQTT-C/blob/master/LICENSE"><img src="https://img.shields.io/badge/License-MIT-blue.svg"></a>
</p>
#
MQTT-C is an [MQTT v3.1.1](http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html)
client written in C. MQTT is a lightweight publisher-subscriber-based messaging protocol that is
commonly used in IoT and networking applications where high-latency and low data-rate links
are expected. The purpose of MQTT-C is to provide a **portable** MQTT client, **written in C**,
for embedded systems and PC's alike. MQTT-C does this by providing a transparent Platform
Abstraction Layer (PAL) which makes porting to new platforms easy. MQTT-C is completely
thread-safe but can also run perfectly fine on single-threaded systems making MQTT-C
well-suited for embedded systems and microcontrollers. Finally, MQTT-C is small; there are only
two source files totalling less than 2000 lines.
#### A note from the author
It's been great to hear about all the places MQTT-C is being used! Please don't hesitate
to get in touch with me or submit issues on GitHub!
## Getting Started
To use MQTT-C you first instantiate a `struct mqtt_client` and initialize it by calling
@ref mqtt_init.
```c
struct mqtt_client client; /* instantiate the client */
mqtt_init(&client, ...); /* initialize the client */
```
Once your client is initialized you need to connect to an MQTT broker.
```c
mqtt_connect(&client, ...); /* send a connection request to the broker. */
```
At this point the client is ready to use! For example, we can subscribe to a topic like so:
```c
/* subscribe to "toaster/temperature" with a max QoS level of 0 */
mqtt_subscribe(&client, "toaster/temperature", 0);
```
And we can publish to a topic like so:
```c
/* publish coffee temperature with a QoS level of 1 */
int temperature = 67;
mqtt_publish(&client, "coffee/temperature", &temperature, sizeof(int), MQTT_PUBLISH_QOS_1);
```
Those are the basics! From here the [examples](https://github.com/LiamBindle/MQTT-C/tree/master/examples) and [API documentation](https://liambindle.ca/MQTT-C/group__api.html) are good places to get started.
## Building
There are **only two source files** that need to be built, `mqtt.c` and `mqtt_pal.c`.
These files are ANSI C (C89) compatible, and should compile with any C compiler.
Then, simply <code>\#include <mqtt.h></code>.
Alternatively, you can build MQTT-C with CMake or the provided Makefile. These are provided for convenience.
## Documentation
Pre-built documentation can be found here: [https://liambindle.ca/MQTT-C](https://liambindle.ca/MQTT-C). Be sure to check out the [examples](https://github.com/LiamBindle/MQTT-C/tree/master/examples) too.
The @ref api documentation contains all the documentation application programmers should need.
The @ref pal documentation contains everything you should need to port MQTT-C to a new platform,
and the other modules contain documentation for MQTT-C developers.
## Testing and Building the Tests
The MQTT-C unit tests use the [cmocka unit testing framework](https://cmocka.org/).
Therefore, [cmocka](https://cmocka.org/) *must* be installed on your machine to build and run
the unit tests. For convenience, a simple `"makefile"` is included to build the unit tests and
examples on UNIX-like machines. The unit tests and examples can be built as follows:
```bash
$ make all
```
The unit tests and examples will be built in the `"bin/"` directory. The unit tests can be run
like so:
```bash
$ ./bin/tests [address [port]]
```
Note that the \c address and \c port arguments are both optional to specify the location of the
MQTT broker that is to be used for the tests. If no \c address is given then the
[Mosquitto MQTT Test Server](https://test.mosquitto.org/) will be used. If no \c port is given,
port 1883 will be used.
## Portability
MQTT-C provides a transparent platform abstraction layer (PAL) in `mqtt_pal.h` and `mqtt_pal.c`.
These files declare and implement the types and calls that MQTT-C requires. Refer to
@ref pal for the complete documentation of the PAL.
## Contributing
Please feel free to submit issues and pull-requests [here](https://github.com/LiamBindle/MQTT-C).
When submitting a pull-request please ensure you have *fully documented* your changes and
added the appropriate unit tests.
## License
This project is licensed under the [MIT License](https://opensource.org/licenses/MIT). See the
`"LICENSE"` file for more details.
## Authors
MQTT-C was initially developed as a CMPT 434 (Winter Term, 2018) final project at the University of
Saskatchewan by:
- **Liam Bindle**
- **Demilade Adeoye**
deps/MQTT-C/examples/bio_publisher.c 0 → 100644
浏览文件 @ 3f64b14a
/**
* @file
* A simple program to that publishes the current time whenever ENTER is pressed.
*/
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <mqtt.h>
#include "templates/bio_sockets.h"
/**
* @brief The function that would be called whenever a PUBLISH is received.
*
* @note This function is not used in this example.
*/
void publish_callback(void** unused, struct mqtt_response_publish *published);
/**
* @brief The client's refresher. This function triggers back-end routines to
* handle ingress/egress traffic to the broker.
*
* @note All this function needs to do is call \ref __mqtt_recv and
* \ref __mqtt_send every so often. I've picked 100 ms meaning that
* client ingress/egress traffic will be handled every 100 ms.
*/
void* client_refresher(void* client);
/**
* @brief Safelty closes the \p sockfd and cancels the \p client_daemon before \c exit.
*/
void exit_example(int status, BIO* sockfd, pthread_t *client_daemon);
/**
* A simple program to that publishes the current time whenever ENTER is pressed.
*/
int main(int argc, const char *argv[])
{
const char* addr;
const char* port;
const char* topic;
/* Load OpenSSL */
SSL_load_error_strings();
ERR_load_BIO_strings();
OpenSSL_add_all_algorithms();
/* get address (argv[1] if present) */
if (argc > 1) {
addr = argv[1];
} else {
addr = "test.mosquitto.org";
}
/* get port number (argv[2] if present) */
if (argc > 2) {
port = argv[2];
} else {
port = "1883";
}
/* get the topic name to publish */
if (argc > 3) {
topic = argv[3];
} else {
topic = "datetime";
}
/* open the non-blocking TCP socket (connecting to the broker) */
BIO* sockfd = open_nb_socket(addr, port);
if (sockfd == NULL) {
exit_example(EXIT_FAILURE, sockfd, NULL);
}
/* setup a client */
struct mqtt_client client;
uint8_t sendbuf[2048]; /* sendbuf should be large enough to hold multiple whole mqtt messages */
uint8_t recvbuf[1024]; /* recvbuf should be large enough any whole mqtt message expected to be received */
mqtt_init(&client, sockfd, sendbuf, sizeof(sendbuf), recvbuf, sizeof(recvbuf), publish_callback);
mqtt_connect(&client, "publishing_client", NULL, NULL, 0, NULL, NULL, 0, 400);
/* check that we don't have any errors */
if (client.error != MQTT_OK) {
fprintf(stderr, "error: %s\n", mqtt_error_str(client.error));
exit_example(EXIT_FAILURE, sockfd, NULL);
}
/* start a thread to refresh the client (handle egress and ingree client traffic) */
pthread_t client_daemon;
if(pthread_create(&client_daemon, NULL, client_refresher, &client)) {
fprintf(stderr, "Failed to start client daemon.\n");
exit_example(EXIT_FAILURE, sockfd, NULL);
}
/* start publishing the time */
printf("%s is ready to begin publishing the time.\n", argv[0]);
printf("Press ENTER to publish the current time.\n");
printf("Press CTRL-D (or any other key) to exit.\n\n");
while(fgetc(stdin) == '\n') {
/* get the current time */
time_t timer;
time(&timer);
struct tm* tm_info = localtime(&timer);
char timebuf[26];
strftime(timebuf, 26, "%Y-%m-%d %H:%M:%S", tm_info);
/* print a message */
char application_message[256];
snprintf(application_message, sizeof(application_message), "The time is %s", timebuf);
printf("%s published : \"%s\"", argv[0], application_message);
/* publish the time */
mqtt_publish(&client, topic, application_message, strlen(application_message) + 1, MQTT_PUBLISH_QOS_2);
/* check for errors */
if (client.error != MQTT_OK) {
fprintf(stderr, "error: %s\n", mqtt_error_str(client.error));
exit_example(EXIT_FAILURE, sockfd, &client_daemon);
}
}
/* disconnect */
printf("\n%s disconnecting from %s\n", argv[0], addr);
sleep(1);
/* exit */
exit_example(EXIT_SUCCESS, sockfd, &client_daemon);
}
void exit_example(int status, BIO* sockfd, pthread_t *client_daemon)
{
if (sockfd != NULL) BIO_free_all(sockfd);
if (client_daemon != NULL) pthread_cancel(*client_daemon);
exit(status);
}
void publish_callback(void** unused, struct mqtt_response_publish *published)
{
/* not used in this example */
}
void* client_refresher(void* client)
{
while(1)
{
mqtt_sync((struct mqtt_client*) client);
usleep(100000U);
}
return NULL;
}
\ No newline at end of file
deps/MQTT-C/examples/mbedtls_publisher.c 0 → 100644
浏览文件 @ 3f64b14a
/**
* @file
*/
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <mqtt.h>
#include "templates/mbedtls_sockets.h"
/**
* @brief The function that would be called whenever a PUBLISH is received.
*
* @note This function is not used in this example.
*/
void publish_callback(void** unused, struct mqtt_response_publish *published);
/**
* @brief The client's refresher. This function triggers back-end routines to
* handle ingress/egress traffic to the broker.
*
* @note All this function needs to do is call \ref __mqtt_recv and
* \ref __mqtt_send every so often. I've picked 100 ms meaning that
* client ingress/egress traffic will be handled every 100 ms.
*/
void* client_refresher(void* client);
/**
* @brief Safelty closes the \p sockfd and cancels the \p client_daemon before \c exit.
*/
void exit_example(int status, mqtt_pal_socket_handle sockfd, pthread_t *client_daemon);
/**
* A simple program to that publishes the current time whenever ENTER is pressed.
*/
int main(int argc, const char *argv[])
{
const char* addr;
const char* port;
const char* topic;
const char* ca_file;
struct mbedtls_context ctx;
mqtt_pal_socket_handle sockfd;
if (argc > 1) {
ca_file = argv[1];
} else {
printf("error: path to the CA certificate to use\n");
exit(1);
}
/* get address (argv[2] if present) */
if (argc > 2) {
addr = argv[2];
} else {
addr = "test.mosquitto.org";
}
/* get port number (argv[3] if present) */
if (argc > 3) {
port = argv[3];
} else {
port = "8883";
}
/* get the topic name to publish */
if (argc > 4) {
topic = argv[4];
} else {
topic = "datetime";
}
/* open the non-blocking TCP socket (connecting to the broker) */
open_nb_socket(&ctx, addr, port, ca_file);
sockfd = &ctx.ssl_ctx;
if (sockfd == NULL) {
exit_example(EXIT_FAILURE, sockfd, NULL);
}
/* setup a client */
struct mqtt_client client;
uint8_t sendbuf[2048]; /* sendbuf should be large enough to hold multiple whole mqtt messages */
uint8_t recvbuf[1024]; /* recvbuf should be large enough any whole mqtt message expected to be received */
mqtt_init(&client, sockfd, sendbuf, sizeof(sendbuf), recvbuf, sizeof(recvbuf), publish_callback);
mqtt_connect(&client, "publishing_client", NULL, NULL, 0, NULL, NULL, 0, 400);
/* check that we don't have any errors */
if (client.error != MQTT_OK) {
fprintf(stderr, "error: %s\n", mqtt_error_str(client.error));
exit_example(EXIT_FAILURE, sockfd, NULL);
}
/* start a thread to refresh the client (handle egress and ingree client traffic) */
pthread_t client_daemon;
if(pthread_create(&client_daemon, NULL, client_refresher, &client)) {
fprintf(stderr, "Failed to start client daemon.\n");
exit_example(EXIT_FAILURE, sockfd, NULL);
}
/* start publishing the time */
printf("%s is ready to begin publishing the time.\n", argv[0]);
printf("Press ENTER to publish the current time.\n");
printf("Press CTRL-D (or any other key) to exit.\n\n");
while(fgetc(stdin) == '\n') {
/* get the current time */
time_t timer;
time(&timer);
struct tm* tm_info = localtime(&timer);
char timebuf[26];
strftime(timebuf, 26, "%Y-%m-%d %H:%M:%S", tm_info);
/* print a message */
char application_message[256];
snprintf(application_message, sizeof(application_message), "The time is %s", timebuf);
printf("%s published : \"%s\"", argv[0], application_message);
/* publish the time */
mqtt_publish(&client, topic, application_message, strlen(application_message) + 1, MQTT_PUBLISH_QOS_2);
/* check for errors */
if (client.error != MQTT_OK) {
fprintf(stderr, "error: %s\n", mqtt_error_str(client.error));
exit_example(EXIT_FAILURE, sockfd, &client_daemon);
}
}
/* disconnect */
printf("\n%s disconnecting from %s\n", argv[0], addr);
sleep(1);
/* exit */
exit_example(EXIT_SUCCESS, sockfd, &client_daemon);
}
void exit_example(int status, mqtt_pal_socket_handle sockfd, pthread_t *client_daemon)
{
if (client_daemon != NULL) pthread_cancel(*client_daemon);
mbedtls_ssl_free(sockfd);
/* XXX free the rest of contexts */
exit(status);
}
void publish_callback(void** unused, struct mqtt_response_publish *published)
{
/* not used in this example */
}
void* client_refresher(void* client)
{
while(1)
{
mqtt_sync((struct mqtt_client*) client);
usleep(100000U);
}
return NULL;
}
deps/MQTT-C/examples/openssl_publisher.c 0 → 100644
浏览文件 @ 3f64b14a
/**
* @file
*/
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <mqtt.h>
#include "templates/openssl_sockets.h"
/**
* @brief The function that would be called whenever a PUBLISH is received.
*
* @note This function is not used in this example.
*/
void publish_callback(void** unused, struct mqtt_response_publish *published);
/**
* @brief The client's refresher. This function triggers back-end routines to
* handle ingress/egress traffic to the broker.
*
* @note All this function needs to do is call \ref __mqtt_recv and
* \ref __mqtt_send every so often. I've picked 100 ms meaning that
* client ingress/egress traffic will be handled every 100 ms.
*/
void* client_refresher(void* client);
/**
* @brief Safelty closes the \p sockfd and cancels the \p client_daemon before \c exit.
*/
void exit_example(int status, BIO* sockfd, pthread_t *client_daemon);
/**
* A simple program to that publishes the current time whenever ENTER is pressed.
*/
int main(int argc, const char *argv[])
{
const char* addr;
const char* port;
const char* topic;
const char* ca_file;
/* Load OpenSSL */
SSL_load_error_strings();
ERR_load_BIO_strings();
OpenSSL_add_all_algorithms();
SSL_library_init();
SSL_CTX* ssl_ctx;
BIO* sockfd;
if (argc > 1) {
ca_file = argv[1];
} else {
printf("error: path to the CA certificate to use\n");
exit(1);
}
/* get address (argv[2] if present) */
if (argc > 2) {
addr = argv[2];
} else {
addr = "test.mosquitto.org";
}
/* get port number (argv[3] if present) */
if (argc > 3) {
port = argv[3];
} else {
port = "8883";
}
/* get the topic name to publish */
if (argc > 4) {
topic = argv[4];
} else {
topic = "datetime";
}
/* open the non-blocking TCP socket (connecting to the broker) */
open_nb_socket(&sockfd, &ssl_ctx, addr, port, ca_file, NULL);
if (sockfd == NULL) {
exit_example(EXIT_FAILURE, sockfd, NULL);
}
/* setup a client */
struct mqtt_client client;
uint8_t sendbuf[2048]; /* sendbuf should be large enough to hold multiple whole mqtt messages */
uint8_t recvbuf[1024]; /* recvbuf should be large enough any whole mqtt message expected to be received */
mqtt_init(&client, sockfd, sendbuf, sizeof(sendbuf), recvbuf, sizeof(recvbuf), publish_callback);
mqtt_connect(&client, "publishing_client", NULL, NULL, 0, NULL, NULL, 0, 400);
/* check that we don't have any errors */
if (client.error != MQTT_OK) {
fprintf(stderr, "error: %s\n", mqtt_error_str(client.error));
exit_example(EXIT_FAILURE, sockfd, NULL);
}
/* start a thread to refresh the client (handle egress and ingree client traffic) */
pthread_t client_daemon;
if(pthread_create(&client_daemon, NULL, client_refresher, &client)) {
fprintf(stderr, "Failed to start client daemon.\n");
exit_example(EXIT_FAILURE, sockfd, NULL);
}
/* start publishing the time */
printf("%s is ready to begin publishing the time.\n", argv[0]);
printf("Press ENTER to publish the current time.\n");
printf("Press CTRL-D (or any other key) to exit.\n\n");
while(fgetc(stdin) == '\n') {
/* get the current time */
time_t timer;
time(&timer);
struct tm* tm_info = localtime(&timer);
char timebuf[26];
strftime(timebuf, 26, "%Y-%m-%d %H:%M:%S", tm_info);
/* print a message */
char application_message[256];
snprintf(application_message, sizeof(application_message), "The time is %s", timebuf);
printf("%s published : \"%s\"", argv[0], application_message);
/* publish the time */
mqtt_publish(&client, topic, application_message, strlen(application_message) + 1, MQTT_PUBLISH_QOS_2);
/* check for errors */
if (client.error != MQTT_OK) {
fprintf(stderr, "error: %s\n", mqtt_error_str(client.error));
exit_example(EXIT_FAILURE, sockfd, &client_daemon);
}
}
/* disconnect */
printf("\n%s disconnecting from %s\n", argv[0], addr);
sleep(1);
/* exit */
exit_example(EXIT_SUCCESS, sockfd, &client_daemon);
}
void exit_example(int status, BIO* sockfd, pthread_t *client_daemon)
{
if (sockfd != NULL) BIO_free_all(sockfd);
if (client_daemon != NULL) pthread_cancel(*client_daemon);
exit(status);
}
void publish_callback(void** unused, struct mqtt_response_publish *published)
{
/* not used in this example */
}
void* client_refresher(void* client)
{
while(1)
{
mqtt_sync((struct mqtt_client*) client);
usleep(100000U);
}
return NULL;
}
\ No newline at end of file
deps/MQTT-C/examples/reconnect_subscriber.c 0 → 100644
浏览文件 @ 3f64b14a
/**
* @file
* A simple subscriber program that performs automatic reconnections.
*/
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <mqtt.h>
#include "templates/posix_sockets.h"
/**
* @brief A structure that I will use to keep track of some data needed
* to setup the connection to the broker.
*
* An instance of this struct will be created in my \c main(). Then, whenever
* \ref reconnect_client is called, this instance will be passed.
*/
struct reconnect_state_t {
const char* hostname;
const char* port;
const char* topic;
uint8_t* sendbuf;
size_t sendbufsz;
uint8_t* recvbuf;
size_t recvbufsz;
};
/**
* @brief My reconnect callback. It will reestablish the connection whenever
* an error occurs.
*/
void reconnect_client(struct mqtt_client* client, void **reconnect_state_vptr);
/**
* @brief The function will be called whenever a PUBLISH message is received.
*/
void publish_callback(void** unused, struct mqtt_response_publish *published);
/**
* @brief The client's refresher. This function triggers back-end routines to
* handle ingress/egress traffic to the broker.
*
* @note All this function needs to do is call \ref __mqtt_recv and
* \ref __mqtt_send every so often. I've picked 100 ms meaning that
* client ingress/egress traffic will be handled every 100 ms.
*/
void* client_refresher(void* client);
/**
* @brief Safelty closes the \p sockfd and cancels the \p client_daemon before \c exit.
*/
void exit_example(int status, int sockfd, pthread_t *client_daemon);
int main(int argc, const char *argv[])
{
const char* addr;
const char* port;
const char* topic;
/* get address (argv[1] if present) */
if (argc > 1) {
addr = argv[1];
} else {
addr = "test.mosquitto.org";
}
/* get port number (argv[2] if present) */
if (argc > 2) {
port = argv[2];
} else {
port = "1883";
}
/* get the topic name to publish */
if (argc > 3) {
topic = argv[3];
} else {
topic = "datetime";
}
/* build the reconnect_state structure which will be passed to reconnect */
struct reconnect_state_t reconnect_state;
reconnect_state.hostname = addr;
reconnect_state.port = port;
reconnect_state.topic = topic;
uint8_t sendbuf[2048];
uint8_t recvbuf[1024];
reconnect_state.sendbuf = sendbuf;
reconnect_state.sendbufsz = sizeof(sendbuf);
reconnect_state.recvbuf = recvbuf;
reconnect_state.recvbufsz = sizeof(recvbuf);
/* setup a client */
struct mqtt_client client;
mqtt_init_reconnect(&client,
reconnect_client, &reconnect_state,
publish_callback
);
/* start a thread to refresh the client (handle egress and ingree client traffic) */
pthread_t client_daemon;
if(pthread_create(&client_daemon, NULL, client_refresher, &client)) {
fprintf(stderr, "Failed to start client daemon.\n");
exit_example(EXIT_FAILURE, -1, NULL);
}
/* start publishing the time */
printf("%s listening for '%s' messages.\n", argv[0], topic);
printf("Press ENTER to inject an error.\n");
printf("Press CTRL-D to exit.\n\n");
/* block */
while(fgetc(stdin) != EOF) {
printf("Injecting error: \"MQTT_ERROR_SOCKET_ERROR\"\n");
client.error = MQTT_ERROR_SOCKET_ERROR;
}
/* disconnect */
printf("\n%s disconnecting from %s\n", argv[0], addr);
sleep(1);
/* exit */
exit_example(EXIT_SUCCESS, client.socketfd, &client_daemon);
}
void reconnect_client(struct mqtt_client* client, void **reconnect_state_vptr)
{
struct reconnect_state_t *reconnect_state = *((struct reconnect_state_t**) reconnect_state_vptr);
/* Close the clients socket if this isn't the initial reconnect call */
if (client->error != MQTT_ERROR_INITIAL_RECONNECT) {
close(client->socketfd);
}
/* Perform error handling here. */
if (client->error != MQTT_ERROR_INITIAL_RECONNECT) {
printf("reconnect_client: called while client was in error state \"%s\"\n",
mqtt_error_str(client->error)
);
}
/* Open a new socket. */
int sockfd = open_nb_socket(reconnect_state->hostname, reconnect_state->port);
if (sockfd == -1) {
perror("Failed to open socket: ");
exit_example(EXIT_FAILURE, sockfd, NULL);
}
/* Reinitialize the client. */
mqtt_reinit(client, sockfd,
reconnect_state->sendbuf, reconnect_state->sendbufsz,
reconnect_state->recvbuf, reconnect_state->recvbufsz
);
/* Create an anonymous session */
const char* client_id = NULL;
/* Ensure we have a clean session */
uint8_t connect_flags = MQTT_CONNECT_CLEAN_SESSION;
/* Send connection request to the broker. */
mqtt_connect(client, client_id, NULL, NULL, 0, NULL, NULL, connect_flags, 400);
/* Subscribe to the topic. */
mqtt_subscribe(client, reconnect_state->topic, 0);
}
void exit_example(int status, int sockfd, pthread_t *client_daemon)
{
if (sockfd != -1) close(sockfd);
if (client_daemon != NULL) pthread_cancel(*client_daemon);
exit(status);
}
void publish_callback(void** unused, struct mqtt_response_publish *published)
{
/* note that published->topic_name is NOT null-terminated (here we'll change it to a c-string) */
char* topic_name = (char*) malloc(published->topic_name_size + 1);
memcpy(topic_name, published->topic_name, published->topic_name_size);
topic_name[published->topic_name_size] = '\0';
printf("Received publish('%s'): %s\n", topic_name, (const char*) published->application_message);
free(topic_name);
}
void* client_refresher(void* client)
{
while(1)
{
mqtt_sync((struct mqtt_client*) client);
usleep(100000U);
}
return NULL;
}
\ No newline at end of file
deps/MQTT-C/examples/simple_publisher.c 0 → 100644
浏览文件 @ 3f64b14a
/**
* @file
* A simple program to that publishes the current time whenever ENTER is pressed.
*/
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <mqtt.h>
#include "templates/posix_sockets.h"
/**
* @brief The function that would be called whenever a PUBLISH is received.
*
* @note This function is not used in this example.
*/
void publish_callback(void** unused, struct mqtt_response_publish *published);
/**
* @brief The client's refresher. This function triggers back-end routines to
* handle ingress/egress traffic to the broker.
*
* @note All this function needs to do is call \ref __mqtt_recv and
* \ref __mqtt_send every so often. I've picked 100 ms meaning that
* client ingress/egress traffic will be handled every 100 ms.
*/
void* client_refresher(void* client);
/**
* @brief Safelty closes the \p sockfd and cancels the \p client_daemon before \c exit.
*/
void exit_example(int status, int sockfd, pthread_t *client_daemon);
/**
* A simple program to that publishes the current time whenever ENTER is pressed.
*/
int main(int argc, const char *argv[])
{
const char* addr;
const char* port;
const char* topic;
/* get address (argv[1] if present) */
if (argc > 1) {
addr = argv[1];
} else {
addr = "test.mosquitto.org";
}
/* get port number (argv[2] if present) */
if (argc > 2) {
port = argv[2];
} else {
port = "1883";
}
/* get the topic name to publish */
if (argc > 3) {
topic = argv[3];
} else {
topic = "datetime";
}
/* open the non-blocking TCP socket (connecting to the broker) */
int sockfd = open_nb_socket(addr, port);
if (sockfd == -1) {
perror("Failed to open socket: ");
exit_example(EXIT_FAILURE, sockfd, NULL);
}
/* setup a client */
struct mqtt_client client;
uint8_t sendbuf[2048]; /* sendbuf should be large enough to hold multiple whole mqtt messages */
uint8_t recvbuf[1024]; /* recvbuf should be large enough any whole mqtt message expected to be received */
mqtt_init(&client, sockfd, sendbuf, sizeof(sendbuf), recvbuf, sizeof(recvbuf), publish_callback);
/* Create an anonymous session */
const char* client_id = NULL;
/* Ensure we have a clean session */
uint8_t connect_flags = MQTT_CONNECT_CLEAN_SESSION;
/* Send connection request to the broker. */
mqtt_connect(&client, client_id, NULL, NULL, 0, NULL, NULL, connect_flags, 400);
/* check that we don't have any errors */
if (client.error != MQTT_OK) {
fprintf(stderr, "error: %s\n", mqtt_error_str(client.error));
exit_example(EXIT_FAILURE, sockfd, NULL);
}
/* start a thread to refresh the client (handle egress and ingree client traffic) */
pthread_t client_daemon;
if(pthread_create(&client_daemon, NULL, client_refresher, &client)) {
fprintf(stderr, "Failed to start client daemon.\n");
exit_example(EXIT_FAILURE, sockfd, NULL);
}
/* start publishing the time */
printf("%s is ready to begin publishing the time.\n", argv[0]);
printf("Press ENTER to publish the current time.\n");
printf("Press CTRL-D (or any other key) to exit.\n\n");
while(fgetc(stdin) == '\n') {
/* get the current time */
time_t timer;
time(&timer);
struct tm* tm_info = localtime(&timer);
char timebuf[26];
strftime(timebuf, 26, "%Y-%m-%d %H:%M:%S", tm_info);
/* print a message */
char application_message[256];
snprintf(application_message, sizeof(application_message), "The time is %s", timebuf);
printf("%s published : \"%s\"", argv[0], application_message);
/* publish the time */
mqtt_publish(&client, topic, application_message, strlen(application_message) + 1, MQTT_PUBLISH_QOS_0);
/* check for errors */
if (client.error != MQTT_OK) {
fprintf(stderr, "error: %s\n", mqtt_error_str(client.error));
exit_example(EXIT_FAILURE, sockfd, &client_daemon);
}
}
/* disconnect */
printf("\n%s disconnecting from %s\n", argv[0], addr);
sleep(1);
/* exit */
exit_example(EXIT_SUCCESS, sockfd, &client_daemon);
}
void exit_example(int status, int sockfd, pthread_t *client_daemon)
{
if (sockfd != -1) close(sockfd);
if (client_daemon != NULL) pthread_cancel(*client_daemon);
exit(status);
}
void publish_callback(void** unused, struct mqtt_response_publish *published)
{
/* not used in this example */
}
void* client_refresher(void* client)
{
while(1)
{
mqtt_sync((struct mqtt_client*) client);
usleep(100000U);
}
return NULL;
}
\ No newline at end of file
deps/MQTT-C/examples/simple_subscriber.c 0 → 100644
浏览文件 @ 3f64b14a
/**
* @file
* A simple program that subscribes to a topic.
*/
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <mqtt.h>
#include "templates/posix_sockets.h"
/**
* @brief The function will be called whenever a PUBLISH message is received.
*/
void publish_callback(void** unused, struct mqtt_response_publish *published);
/**
* @brief The client's refresher. This function triggers back-end routines to
* handle ingress/egress traffic to the broker.
*
* @note All this function needs to do is call \ref __mqtt_recv and
* \ref __mqtt_send every so often. I've picked 100 ms meaning that
* client ingress/egress traffic will be handled every 100 ms.
*/
void* client_refresher(void* client);
/**
* @brief Safelty closes the \p sockfd and cancels the \p client_daemon before \c exit.
*/
void exit_example(int status, int sockfd, pthread_t *client_daemon);
int main(int argc, const char *argv[])
{
const char* addr;
const char* port;
const char* topic;
/* get address (argv[1] if present) */
if (argc > 1) {
addr = argv[1];
} else {
addr = "test.mosquitto.org";
}
/* get port number (argv[2] if present) */
if (argc > 2) {
port = argv[2];
} else {
port = "1883";
}
/* get the topic name to publish */
if (argc > 3) {
topic = argv[3];
} else {
topic = "datetime";
}
/* open the non-blocking TCP socket (connecting to the broker) */
int sockfd = open_nb_socket(addr, port);
if (sockfd == -1) {
perror("Failed to open socket: ");
exit_example(EXIT_FAILURE, sockfd, NULL);
}
/* setup a client */
struct mqtt_client client;
uint8_t sendbuf[2048]; /* sendbuf should be large enough to hold multiple whole mqtt messages */
uint8_t recvbuf[1024]; /* recvbuf should be large enough any whole mqtt message expected to be received */
mqtt_init(&client, sockfd, sendbuf, sizeof(sendbuf), recvbuf, sizeof(recvbuf), publish_callback);
/* Create an anonymous session */
const char* client_id = NULL;
/* Ensure we have a clean session */
uint8_t connect_flags = MQTT_CONNECT_CLEAN_SESSION;
/* Send connection request to the broker. */
mqtt_connect(&client, client_id, NULL, NULL, 0, NULL, NULL, connect_flags, 400);
/* check that we don't have any errors */
if (client.error != MQTT_OK) {
fprintf(stderr, "error: %s\n", mqtt_error_str(client.error));
exit_example(EXIT_FAILURE, sockfd, NULL);
}
/* start a thread to refresh the client (handle egress and ingree client traffic) */
pthread_t client_daemon;
if(pthread_create(&client_daemon, NULL, client_refresher, &client)) {
fprintf(stderr, "Failed to start client daemon.\n");
exit_example(EXIT_FAILURE, sockfd, NULL);
}
/* subscribe */
mqtt_subscribe(&client, topic, 0);
/* start publishing the time */
printf("%s listening for '%s' messages.\n", argv[0], topic);
printf("Press CTRL-D to exit.\n\n");
/* block */
while(fgetc(stdin) != EOF);
/* disconnect */
printf("\n%s disconnecting from %s\n", argv[0], addr);
sleep(1);
/* exit */
exit_example(EXIT_SUCCESS, sockfd, &client_daemon);
}
void exit_example(int status, int sockfd, pthread_t *client_daemon)
{
if (sockfd != -1) close(sockfd);
if (client_daemon != NULL) pthread_cancel(*client_daemon);
exit(status);
}
void publish_callback(void** unused, struct mqtt_response_publish *published)
{
/* note that published->topic_name is NOT null-terminated (here we'll change it to a c-string) */
char* topic_name = (char*) malloc(published->topic_name_size + 1);
memcpy(topic_name, published->topic_name, published->topic_name_size);
topic_name[published->topic_name_size] = '\0';
printf("Received publish('%s'): %s\n", topic_name, (const char*) published->application_message);
free(topic_name);
}
void* client_refresher(void* client)
{
while(1)
{
mqtt_sync((struct mqtt_client*) client);
usleep(100000U);
}
return NULL;
}
\ No newline at end of file
deps/MQTT-C/examples/templates/bio_sockets.h 0 → 100644
浏览文件 @ 3f64b14a
#ifndef __BIO_SOCKET_TEMPLATE_H__
#define __BIO_SOCKET_TEMPLATE_H__
#include <openssl/bio.h>
#include <openssl/ssl.h>
#include <openssl/err.h>
/*
A template for opening a non-blocking BIO socket.
*/
BIO* open_nb_socket(const char* addr, const char* port) {
BIO* bio = BIO_new_connect(addr);
BIO_set_nbio(bio, 1);
BIO_set_conn_port(bio, port);
/* timeout after 10 seconds */
int start_time = time(NULL);
while(BIO_do_connect(bio) == 0 && (int)time(NULL) - start_time < 10);
if (BIO_do_connect(bio) <= 0) {
fprintf(stderr, "Failed to open socket: BIO_do_connect returned <= 0\n");
return NULL;
}
return bio;
}
#endif
\ No newline at end of file
deps/MQTT-C/examples/templates/mbedtls_sockets.h 0 → 100644
浏览文件 @ 3f64b14a
#ifndef __MBEDTLS_SOCKET_TEMPLATE_H__
#define __MBEDTLS_SOCKET_TEMPLATE_H__
#include <inttypes.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <mbedtls/error.h>
#include <mbedtls/entropy.h>
#include <mbedtls/ctr_drbg.h>
#include <mbedtls/net_sockets.h>
#include <mbedtls/ssl.h>
#if !defined(MBEDTLS_NET_POLL_READ)
/* compat for older mbedtls */
#define MBEDTLS_NET_POLL_READ 1
#define MBEDTLS_NET_POLL_WRITE 1
int
mbedtls_net_poll(mbedtls_net_context * ctx, uint32_t rw, uint32_t timeout)
{
/* XXX this is not ideal but good enough for an example */
usleep(300);
return 1;
}
#endif
struct mbedtls_context {
mbedtls_net_context net_ctx;
mbedtls_ssl_context ssl_ctx;
mbedtls_ssl_config ssl_conf;
mbedtls_x509_crt ca_crt;
mbedtls_entropy_context entropy;
mbedtls_ctr_drbg_context ctr_drbg;
};
void failed(const char *fn, int rv) {
char buf[100];
mbedtls_strerror(rv, buf, sizeof(buf));
printf("%s failed with %x (%s)\n", fn, -rv, buf);
exit(1);
}
void cert_verify_failed(uint32_t rv) {
char buf[512];
mbedtls_x509_crt_verify_info(buf, sizeof(buf), "\t", rv);
printf("Certificate verification failed (%0" PRIx32 ")\n%s\n", rv, buf);
exit(1);
}
/*
A template for opening a non-blocking mbed TLS connection.
*/
void open_nb_socket(struct mbedtls_context *ctx,
const char *hostname,
const char *port,
const char *ca_file) {
const unsigned char *additional = (const unsigned char *)"MQTT-C";
size_t additional_len = 6;
int rv;
mbedtls_net_context *net_ctx = &ctx->net_ctx;
mbedtls_ssl_context *ssl_ctx = &ctx->ssl_ctx;
mbedtls_ssl_config *ssl_conf = &ctx->ssl_conf;
mbedtls_x509_crt *ca_crt = &ctx->ca_crt;
mbedtls_entropy_context *entropy = &ctx->entropy;
mbedtls_ctr_drbg_context *ctr_drbg = &ctx->ctr_drbg;
mbedtls_entropy_init(entropy);
mbedtls_ctr_drbg_init(ctr_drbg);
rv = mbedtls_ctr_drbg_seed(ctr_drbg, mbedtls_entropy_func, entropy,
additional, additional_len);
if (rv != 0) {
failed("mbedtls_ctr_drbg_seed", rv);
}
mbedtls_x509_crt_init(ca_crt);
rv = mbedtls_x509_crt_parse_file(ca_crt, ca_file);
if (rv != 0) {
failed("mbedtls_x509_crt_parse_file", rv);
}
mbedtls_ssl_config_init(ssl_conf);
rv = mbedtls_ssl_config_defaults(ssl_conf, MBEDTLS_SSL_IS_CLIENT,
MBEDTLS_SSL_TRANSPORT_STREAM,
MBEDTLS_SSL_PRESET_DEFAULT);
if (rv != 0) {
failed("mbedtls_ssl_config_defaults", rv);
}
mbedtls_ssl_conf_ca_chain(ssl_conf, ca_crt, NULL);
mbedtls_ssl_conf_authmode(ssl_conf, MBEDTLS_SSL_VERIFY_OPTIONAL);
mbedtls_ssl_conf_rng(ssl_conf, mbedtls_ctr_drbg_random, ctr_drbg);
mbedtls_net_init(net_ctx);
rv = mbedtls_net_connect(net_ctx, hostname, port, MBEDTLS_NET_PROTO_TCP);
if (rv != 0) {
failed("mbedtls_net_connect", rv);
}
rv = mbedtls_net_set_nonblock(net_ctx);
if (rv != 0) {
failed("mbedtls_net_set_nonblock", rv);
}
mbedtls_ssl_init(ssl_ctx);
rv = mbedtls_ssl_setup(ssl_ctx, ssl_conf);
if (rv != 0) {
failed("mbedtls_ssl_setup", rv);
}
rv = mbedtls_ssl_set_hostname(ssl_ctx, hostname);
if (rv != 0) {
failed("mbedtls_ssl_set_hostname", rv);
}
mbedtls_ssl_set_bio(ssl_ctx, net_ctx,
mbedtls_net_send, mbedtls_net_recv, NULL);
for (;;) {
rv = mbedtls_ssl_handshake(ssl_ctx);
uint32_t want = 0;
if (rv == MBEDTLS_ERR_SSL_WANT_READ) {
want |= MBEDTLS_NET_POLL_READ;
} else if (rv == MBEDTLS_ERR_SSL_WANT_WRITE) {
want |= MBEDTLS_NET_POLL_WRITE;
} else {
break;
}
rv = mbedtls_net_poll(net_ctx, want, -1);
if (rv < 0) {
failed("mbedtls_net_poll", rv);
}
}
if (rv != 0) {
failed("mbedtls_ssl_handshake", rv);
}
uint32_t result = mbedtls_ssl_get_verify_result(ssl_ctx);
if (result != 0) {
if (result == (uint32_t)-1) {
failed("mbedtls_ssl_get_verify_result", result);
} else {
cert_verify_failed(result);
}
}
}
#endif
deps/MQTT-C/examples/templates/openssl_sockets.h 0 → 100644
浏览文件 @ 3f64b14a
#ifndef __OPENSSL_SOCKET_TEMPLATE_H__
#define __OPENSSL_SOCKET_TEMPLATE_H__
#include <openssl/bio.h>
#include <openssl/ssl.h>
#include <openssl/err.h>
/*
A template for opening a non-blocking OpenSSL connection.
*/
void open_nb_socket(BIO** bio, SSL_CTX** ssl_ctx, const char* addr, const char* port, const char* ca_file, const char* ca_path) {
*ssl_ctx = SSL_CTX_new(SSLv23_client_method());
SSL* ssl;
/* load certificate */
if (!SSL_CTX_load_verify_locations(*ssl_ctx, ca_file, ca_path)) {
printf("error: failed to load certificate\n");
exit(1);
}
/* open BIO socket */
*bio = BIO_new_ssl_connect(*ssl_ctx);
BIO_get_ssl(*bio, &ssl);
SSL_set_mode(ssl, SSL_MODE_AUTO_RETRY);
BIO_set_conn_hostname(*bio, addr);
BIO_set_nbio(*bio, 1);
BIO_set_conn_port(*bio, port);
/* wait for connect with 10 second timeout */
int start_time = time(NULL);
int do_connect_rv = BIO_do_connect(*bio);
while(do_connect_rv <= 0 && BIO_should_retry(*bio) && (int)time(NULL) - start_time < 10) {
do_connect_rv = BIO_do_connect(*bio);
}
if (do_connect_rv <= 0) {
printf("error: %s\n", ERR_reason_error_string(ERR_get_error()));
BIO_free_all(*bio);
SSL_CTX_free(*ssl_ctx);
*bio = NULL;
*ssl_ctx=NULL;
return;
}
/* verify certificate */
if (SSL_get_verify_result(ssl) != X509_V_OK) {
/* Handle the failed verification */
printf("error: x509 certificate verification failed\n");
exit(1);
}
}
#endif
\ No newline at end of file
deps/MQTT-C/examples/templates/posix_sockets.h 0 → 100644
浏览文件 @ 3f64b14a
#ifndef __POSIX_SOCKET_TEMPLATE_H__
#define __POSIX_SOCKET_TEMPLATE_H__
#include <stdio.h>
#include <sys/types.h>
#if !defined(WIN32)
#include <sys/socket.h>
#include <netdb.h>
#endif
#include <fcntl.h>
/*
A template for opening a non-blocking POSIX socket.
*/
int open_nb_socket(const char* addr, const char* port) {
struct addrinfo hints = {0};
hints.ai_family = AF_UNSPEC; /* IPv4 or IPv6 */
hints.ai_socktype = SOCK_STREAM; /* Must be TCP */
int sockfd = -1;
int rv;
struct addrinfo *p, *servinfo;
/* get address information */
rv = getaddrinfo(addr, port, &hints, &servinfo);
if(rv != 0) {
fprintf(stderr, "Failed to open socket (getaddrinfo): %s\n", gai_strerror(rv));
return -1;
}
/* open the first possible socket */
for(p = servinfo; p != NULL; p = p->ai_next) {
sockfd = socket(p->ai_family, p->ai_socktype, p->ai_protocol);
if (sockfd == -1) continue;
/* connect to server */
rv = connect(sockfd, servinfo->ai_addr, servinfo->ai_addrlen);
if(rv == -1) continue;
break;
}
/* free servinfo */
freeaddrinfo(servinfo);
/* make non-blocking */
#if !defined(WIN32)
if (sockfd != -1) fcntl(sockfd, F_SETFL, fcntl(sockfd, F_GETFL) | O_NONBLOCK);
#else
if (sockfd != INVALID_SOCKET) {
int iMode = 1;
ioctlsocket(sockfd, FIONBIO, &iMode);
}
#endif
/* return the new socket fd */
return sockfd;
}
#endif
\ No newline at end of file
deps/MQTT-C/include/mqtt.h 0 → 100644
浏览文件 @ 3f64b14a
此差异已折叠。
deps/MQTT-C/include/mqtt_pal.h 0 → 100644
浏览文件 @ 3f64b14a
#ifndef __MQTT_PAL_H__
#define __MQTT_PAL_H__
/*
MIT License
Copyright(c) 2018 Liam Bindle
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files(the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions :
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/
/**
* @file
* @brief Includes/supports the types/calls required by the MQTT-C client.
*
* @note This is the \em only file included in mqtt.h, and mqtt.c. It is therefore
* responsible for including/supporting all the required types and calls.
*
* @defgroup pal Platform abstraction layer
* @brief Documentation of the types and calls required to port MQTT-C to a new platform.
*
* mqtt_pal.h is the \em only header file included in mqtt.c. Therefore, to port MQTT-C to a
* new platform the following types, functions, constants, and macros must be defined in
* mqtt_pal.h:
* - Types:
* - \c size_t, \c ssize_t
* - \c uint8_t, \c uint16_t, \c uint32_t
* - \c va_list
* - \c mqtt_pal_time_t : return type of \c MQTT_PAL_TIME()
* - \c mqtt_pal_mutex_t : type of the argument that is passed to \c MQTT_PAL_MUTEX_LOCK and
* \c MQTT_PAL_MUTEX_RELEASE
* - Functions:
* - \c memcpy, \c strlen
* - \c va_start, \c va_arg, \c va_end
* - Constants:
* - \c INT_MIN
*
* Additionally, three macro's are required:
* - \c MQTT_PAL_HTONS(s) : host-to-network endian conversion for uint16_t.
* - \c MQTT_PAL_NTOHS(s) : network-to-host endian conversion for uint16_t.
* - \c MQTT_PAL_TIME() : returns [type: \c mqtt_pal_time_t] current time in seconds.
* - \c MQTT_PAL_MUTEX_LOCK(mtx_pointer) : macro that locks the mutex pointed to by \c mtx_pointer.
* - \c MQTT_PAL_MUTEX_RELEASE(mtx_pointer) : macro that unlocks the mutex pointed to by
* \c mtx_pointer.
*
* Lastly, \ref mqtt_pal_sendall and \ref mqtt_pal_recvall, must be implemented in mqtt_pal.c
* for sending and receiving data using the platforms socket calls.
*/
/* UNIX-like platform support */
#if defined(__unix__) || defined(__APPLE__)
#include <limits.h>
#include <string.h>
#include <stdarg.h>
#include <time.h>
#include <arpa/inet.h>
#include <pthread.h>
#define MQTT_PAL_HTONS(s) htons(s)
#define MQTT_PAL_NTOHS(s) ntohs(s)
#define MQTT_PAL_TIME() time(NULL)
typedef time_t mqtt_pal_time_t;
typedef pthread_mutex_t mqtt_pal_mutex_t;
#define MQTT_PAL_MUTEX_INIT(mtx_ptr) pthread_mutex_init(mtx_ptr, NULL)
#define MQTT_PAL_MUTEX_LOCK(mtx_ptr) pthread_mutex_lock(mtx_ptr)
#define MQTT_PAL_MUTEX_UNLOCK(mtx_ptr) pthread_mutex_unlock(mtx_ptr)
#ifndef MQTT_USE_CUSTOM_SOCKET_HANDLE
#ifdef MQTT_USE_MBEDTLS
struct mbedtls_ssl_context;
typedef struct mbedtls_ssl_context *mqtt_pal_socket_handle;
#elif defined(MQTT_USE_BIO)
#include <openssl/bio.h>
typedef BIO* mqtt_pal_socket_handle;
#else
typedef int mqtt_pal_socket_handle;
#endif
#endif
#elif defined(_MSC_VER)
#include <limits.h>
#include <windows.h>
#include <time.h>
#include <stdint.h>
#include <winsock2.h>
typedef SSIZE_T ssize_t;
#define MQTT_PAL_HTONS(s) htons(s)
#define MQTT_PAL_NTOHS(s) ntohs(s)
#define MQTT_PAL_TIME() time(NULL)
typedef time_t mqtt_pal_time_t;
typedef CRITICAL_SECTION mqtt_pal_mutex_t;
#define MQTT_PAL_MUTEX_INIT(mtx_ptr) InitializeCriticalSection(mtx_ptr)
#define MQTT_PAL_MUTEX_LOCK(mtx_ptr) EnterCriticalSection(mtx_ptr)
#define MQTT_PAL_MUTEX_UNLOCK(mtx_ptr) LeaveCriticalSection(mtx_ptr)
#ifndef MQTT_USE_CUSTOM_SOCKET_HANDLE
#ifdef MQTT_USE_BIO
#include <openssl/bio.h>
typedef BIO* mqtt_pal_socket_handle;
#else
typedef SOCKET mqtt_pal_socket_handle;
#endif
#endif
#endif
/**
* @brief Sends all the bytes in a buffer.
* @ingroup pal
*
* @param[in] fd The file-descriptor (or handle) of the socket.
* @param[in] buf A pointer to the first byte in the buffer to send.
* @param[in] len The number of bytes to send (starting at \p buf).
* @param[in] flags Flags which are passed to the underlying socket.
*
* @returns The number of bytes sent if successful, an \ref MQTTErrors otherwise.
*/
ssize_t mqtt_pal_sendall(mqtt_pal_socket_handle fd, const void* buf, size_t len, int flags);
/**
* @brief Non-blocking receive all the byte available.
* @ingroup pal
*
* @param[in] fd The file-descriptor (or handle) of the socket.
* @param[in] buf A pointer to the receive buffer.
* @param[in] bufsz The max number of bytes that can be put into \p buf.
* @param[in] flags Flags which are passed to the underlying socket.
*
* @returns The number of bytes received if successful, an \ref MQTTErrors otherwise.
*/
ssize_t mqtt_pal_recvall(mqtt_pal_socket_handle fd, void* buf, size_t bufsz, int flags);
#endif
deps/MQTT-C/src/mqtt.c 0 → 100644
浏览文件 @ 3f64b14a
此差异已折叠。
deps/MQTT-C/src/mqtt_pal.c 0 → 100644
浏览文件 @ 3f64b14a
/*
MIT License
Copyright(c) 2018 Liam Bindle
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files(the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions :
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/
#include <mqtt.h>
/**
* @file
* @brief Implements @ref mqtt_pal_sendall and @ref mqtt_pal_recvall and
* any platform-specific helpers you'd like.
* @cond Doxygen_Suppress
*/
#ifdef MQTT_USE_MBEDTLS
#include <mbedtls/ssl.h>
ssize_t mqtt_pal_sendall(mqtt_pal_socket_handle fd, const void* buf, size_t len, int flags) {
size_t sent = 0;
while(sent < len) {
int rv = mbedtls_ssl_write(fd, buf + sent, len - sent);
if (rv < 0) {
if (rv == MBEDTLS_ERR_SSL_WANT_READ ||
rv == MBEDTLS_ERR_SSL_WANT_WRITE
#if defined(MBEDTLS_ERR_SSL_ASYNC_IN_PROGRESS)
|| rv == MBEDTLS_ERR_SSL_ASYNC_IN_PROGRESS
#endif
#if defined(MBEDTLS_ERR_SSL_CRYPTO_IN_PROGRESS)
|| rv == MBEDTLS_ERR_SSL_CRYPTO_IN_PROGRESS
#endif
) {
/* should call mbedtls_ssl_writer later again */
break;
}
return MQTT_ERROR_SOCKET_ERROR;
}
sent += (size_t) rv;
}
return sent;
}
ssize_t mqtt_pal_recvall(mqtt_pal_socket_handle fd, void* buf, size_t bufsz, int flags) {
const void *const start = buf;
int rv;
do {
rv = mbedtls_ssl_read(fd, buf, bufsz);
if (rv < 0) {
if (rv == MBEDTLS_ERR_SSL_WANT_READ ||
rv == MBEDTLS_ERR_SSL_WANT_WRITE
#if defined(MBEDTLS_ERR_SSL_ASYNC_IN_PROGRESS)
|| rv == MBEDTLS_ERR_SSL_ASYNC_IN_PROGRESS
#endif
#if defined(MBEDTLS_ERR_SSL_CRYPTO_IN_PROGRESS)
|| rv == MBEDTLS_ERR_SSL_CRYPTO_IN_PROGRESS
#endif
) {
/* should call mbedtls_ssl_read later again */
break;
}
return MQTT_ERROR_SOCKET_ERROR;
}
buf = (char*)buf + rv;
bufsz -= rv;
} while (rv > 0);
return buf - start;
}
#elif defined(MQTT_USE_BIO)
#include <openssl/bio.h>
#include <openssl/ssl.h>
#include <openssl/err.h>
ssize_t mqtt_pal_sendall(mqtt_pal_socket_handle fd, const void* buf, size_t len, int flags) {
size_t sent = 0;
while(sent < len) {
int tmp = BIO_write(fd, buf + sent, len - sent);
if (tmp > 0) {
sent += (size_t) tmp;
} else if (tmp <= 0 && !BIO_should_retry(fd)) {
return MQTT_ERROR_SOCKET_ERROR;
}
}
return sent;
}
ssize_t mqtt_pal_recvall(mqtt_pal_socket_handle fd, void* buf, size_t bufsz, int flags) {
const void *const start = buf;
int rv;
do {
rv = BIO_read(fd, buf, bufsz);
if (rv > 0) {
/* successfully read bytes from the socket */
buf += rv;
bufsz -= rv;
} else if (!BIO_should_retry(fd)) {
/* an error occurred that wasn't "nothing to read". */
return MQTT_ERROR_SOCKET_ERROR;
}
} while (!BIO_should_read(fd));
return (ssize_t)(buf - start);
}
#elif defined(__unix__) || defined(__APPLE__)
#include <errno.h>
ssize_t mqtt_pal_sendall(mqtt_pal_socket_handle fd, const void* buf, size_t len, int flags) {
size_t sent = 0;
while(sent < len) {
ssize_t tmp = send(fd, buf + sent, len - sent, flags);
if (tmp < 1) {
return MQTT_ERROR_SOCKET_ERROR;
}
sent += (size_t) tmp;
}
return sent;
}
ssize_t mqtt_pal_recvall(mqtt_pal_socket_handle fd, void* buf, size_t bufsz, int flags) {
const void *const start = buf;
ssize_t rv;
do {
rv = recv(fd, buf, bufsz, flags);
if (rv > 0) {
/* successfully read bytes from the socket */
buf += rv;
bufsz -= rv;
} else if (rv < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
/* an error occurred that wasn't "nothing to read". */
return MQTT_ERROR_SOCKET_ERROR;
}
} while (rv > 0);
return buf - start;
}
#elif defined(_MSC_VER)
#include <errno.h>
ssize_t mqtt_pal_sendall(mqtt_pal_socket_handle fd, const void* buf, size_t len, int flags) {
size_t sent = 0;
while(sent < len) {
ssize_t tmp = send(fd, (char*)buf + sent, len - sent, flags);
if (tmp < 1) {
return MQTT_ERROR_SOCKET_ERROR;
}
sent += (size_t) tmp;
}
return sent;
}
ssize_t mqtt_pal_recvall(mqtt_pal_socket_handle fd, void* buf, size_t bufsz, int flags) {
const char *const start = buf;
ssize_t rv;
do {
rv = recv(fd, buf, bufsz, flags);
if (rv > 0) {
/* successfully read bytes from the socket */
buf = (char*)buf + rv;
bufsz -= rv;
} else if (rv < 0) {
int err = WSAGetLastError();
if (err != WSAEWOULDBLOCK) {
/* an error occurred that wasn't "nothing to read". */
return MQTT_ERROR_SOCKET_ERROR;
}
}
} while (rv > 0);
return (ssize_t)((char*)buf - start);
}
#else
#error No PAL!
#endif
/** @endcond */
deps/MQTT-C/tests.c 0 → 100644
浏览文件 @ 3f64b14a
此差异已折叠。
documentation/webdocs/markdowndocs/connector-ch.md
浏览文件 @ 3f64b14a
......@@ -1209,13 +1209,51 @@ TDengine在Window系统上提供的API与Linux系统是相同的， 应用程序
其中，最常用的文件列出如下：
+ Client可执行文件: /usr/local/taos/bin/taos 软连接到 /usr/local/bin/taos
+ 配置文件: /usr/local/taos/cfg/taos.cfg 软连接到 /etc/taos/taos.cfg
+ 驱动程序目录: /usr/local/taos/driver/libtaos.1.6.5.1.dylib 软连接到 /usr/local/lib/libtaos.dylib
+ 驱动程序头文件: /usr/local/taos/include/taos.h 软连接到 /usr/local/include/taos.h
+ 日志目录（第一次运行程序时生成）：~/TDengineLog
## MQTT客户端
MQTT客户端实现了订阅MQTT Broker的特定Topic将Json数据进行转换入库的功能，任何终端只要将数据发给特定的Topic 即可,不用再编写转换器或者数据解析程序。如果终端量大，需要 Mqtt Broker 群集，这里不再详述。
#### 如何配置？
首先需要在 taos.cfg 中打开配置项 mqtt 用来启用， 再通过修改 mqttBrokerAddress 的值来配置连接，格式为:
> mqtt://username:password@hostname:port/path/
例如:
> mqtt://127.0.0.1:1883/taos/ mqtt://root@kissme@127.0.0.1:1883/taos/
#### Topic 格式说明
Mqtt 的topic格式为
> /<path>/<token>/<db name>/<table name>/
因此TDengine的Mqtt客户端会订阅:
> /taos/+/+/+/+/
例如:
> /taos/token/db/t/
注意: 测试时如果需要使用到Mqtt Broker 推荐使用 [mosquitto](http://mosquitto.org/) ，客户端可以使用 [MQTT.fx ](http://www.jensd.de/)
[1]: https://search.maven.org/artifact/com.taosdata.jdbc/taos-jdbcdriver
[2]: https://mvnrepository.com/artifact/com.taosdata.jdbc/taos-jdbcdriver
[3]: https://github.com/taosdata/TDengine
......
packaging/cfg/taos.cfg
浏览文件 @ 3f64b14a
......@@ -166,6 +166,15 @@
# start system monitor module
# monitor 1
# start http service
# mqtt 0
# mqtt uri
# mqttBrokerAddress mqtt://username:password@hostname:1883/taos/
# mqtt client name
# mqttBrokerClientId taos_mqtt
# maximum number of rows returned by the restful interface
# restfulRowLimit 10240
......@@ -244,5 +253,8 @@
# debug flag for system monitor
# monitorDebugFlag 131
#debug flag for mqtt client
# mqttDebugFlag 131
# debug flag for TAOS TIMER
# tmrDebugFlag 131
src/client/inc/tscUtil.h
浏览文件 @ 3f64b14a
......@@ -175,7 +175,7 @@ SSqlExpr* tscSqlExprAppend(SQueryInfo* pQueryInfo, int16_t functionId, SColumnIn
SSqlExpr* tscSqlExprUpdate(SQueryInfo* pQueryInfo, int32_t index, int16_t functionId, int16_t srcColumnIndex, int16_t type,
int16_t size);
int32_t tscSqlExprNumOfExprs(SQueryInfo* pQueryInfo);
size_t tscSqlExprNumOfExprs(SQueryInfo* pQueryInfo);
SSqlExpr* tscSqlExprGet(SQueryInfo* pQueryInfo, int32_t index);
void tscSqlExprCopy(SArray* dst, const SArray* src, uint64_t uid, bool deepcopy);
......
src/client/inc/tsclient.h
浏览文件 @ 3f64b14a
......@@ -84,7 +84,7 @@ typedef struct SSqlExpr {
int16_t functionId; // function id in aAgg array
int16_t resType; // return value type
int16_t resBytes; // length of return value
int16_t interBytes; // inter result buffer size
int32_t interBytes; // inter result buffer size
int16_t numOfParams; // argument value of each function
tVariant param[3]; // parameters are not more than 3
int32_t offset; // sub result column value of arithmetic expression.
......@@ -320,7 +320,7 @@ typedef struct SSqlObj {
tsem_t rspSem;
SSqlCmd cmd;
SSqlRes res;
uint8_t numOfSubs;
uint16_t numOfSubs;
struct SSqlObj **pSubs;
struct SSqlObj * prev, *next;
} SSqlObj;
......
src/client/src/tscAsync.c
浏览文件 @ 3f64b14a
......@@ -57,6 +57,7 @@ void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const
}
pSql->sqlstr = realloc(pSql->sqlstr, sqlLen + 1);
if (pSql->sqlstr == NULL) {
tscError("%p failed to malloc sql string buffer", pSql);
tscQueueAsyncError(fp, param, TSDB_CODE_CLI_OUT_OF_MEMORY);
......@@ -165,7 +166,7 @@ static void tscProcessAsyncRetrieveImpl(void *param, TAOS_RES *tres, int numOfRo
SSqlRes *pRes = &pSql->res;
if ((pRes->qhandle == 0 || numOfRows != 0) && pCmd->command < TSDB_SQL_LOCAL) {
if (pRes->qhandle == 0) {
if (pRes->qhandle == 0 && numOfRows != 0) {
tscError("qhandle is NULL");
} else {
pRes->code = numOfRows;
......
src/client/src/tscFunctionImpl.c
浏览文件 @ 3f64b14a
......@@ -153,7 +153,7 @@ typedef struct SRateInfo {
int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionId, int32_t param, int16_t *type,
int16_t *bytes, int16_t *interBytes, int16_t extLength, bool isSuperTable) {
int16_t *bytes, int32_t *interBytes, int16_t extLength, bool isSuperTable) {
if (!isValidDataType(dataType, dataBytes)) {
tscError("Illegal data type %d or data type length %d", dataType, dataBytes);
return TSDB_CODE_INVALID_SQL;
......@@ -478,7 +478,7 @@ int32_t count_load_data_info(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, int32
if (colId == PRIMARYKEY_TIMESTAMP_COL_INDEX) {
return BLK_DATA_NO_NEEDED;
} else {
return BLK_DATA_FILEDS_NEEDED;
return BLK_DATA_STATIS_NEEDED;
}
}
......@@ -690,7 +690,7 @@ static void sum_func_second_merge(SQLFunctionCtx *pCtx) {
}
static int32_t precal_req_load_info(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, int32_t colId) {
return BLK_DATA_FILEDS_NEEDED;
return BLK_DATA_STATIS_NEEDED;
}
static int32_t data_req_load_info(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, int32_t colId) {
......@@ -1848,13 +1848,14 @@ static void last_row_function(SQLFunctionCtx *pCtx) {
pResInfo->hasResult = DATA_SET_FLAG;
SLastrowInfo *pInfo = (SLastrowInfo *)pResInfo->interResultBuf;
pInfo->ts = pCtx->param[0].i64Key;
pInfo->ts = pCtx->ptsList[0];
pInfo->hasResult = DATA_SET_FLAG;
// set the result to final result buffer
if (pResInfo->superTableQ) {
SLastrowInfo *pInfo1 = (SLastrowInfo *)(pCtx->aOutputBuf + pCtx->inputBytes);
pInfo1->ts = pCtx->param[0].i64Key;
pInfo1->ts = pCtx->ptsList[0];
pInfo1->hasResult = DATA_SET_FLAG;
DO_UPDATE_TAG_COLUMNS(pCtx, pInfo1->ts);
......@@ -1904,12 +1905,12 @@ static void valuePairAssign(tValuePair *dst, int16_t type, const char *val, int6
memcpy(dst->pTags, pTags, (size_t)pTagInfo->tagsLen);
} else { // the tags are dumped from the ctx tag fields
for (int32_t i = 0; i < pTagInfo->numOfTagCols; ++i) {
SQLFunctionCtx* __ctx = pTagInfo->pTagCtxList[i];
if (__ctx->functionId == TSDB_FUNC_TS_DUMMY) {
__ctx->tag = (tVariant) {.nType = TSDB_DATA_TYPE_BIGINT, .i64Key = tsKey};
SQLFunctionCtx* ctx = pTagInfo->pTagCtxList[i];
if (ctx->functionId == TSDB_FUNC_TS_DUMMY) {
ctx->tag = (tVariant) {.nType = TSDB_DATA_TYPE_BIGINT, .i64Key = tsKey};
}
tVariantDump(&pTagInfo->pTagCtxList[i]->tag, dst->pTags + size, pTagInfo->pTagCtxList[i]->tag.nType);
tVariantDump(&ctx->tag, dst->pTags + size, ctx->tag.nType, true);
size += pTagInfo->pTagCtxList[i]->outputBytes;
}
}
......@@ -2226,7 +2227,6 @@ static STopBotInfo *getTopBotOutputInfo(SQLFunctionCtx *pCtx) {
static void buildTopBotStruct(STopBotInfo *pTopBotInfo, SQLFunctionCtx *pCtx) {
char *tmp = (char *)pTopBotInfo + sizeof(STopBotInfo);
pTopBotInfo->res = (tValuePair**) tmp;
tmp += POINTER_BYTES * pCtx->param[0].i64Key;
size_t size = sizeof(tValuePair) + pCtx->tagInfo.tagsLen;
......@@ -2981,14 +2981,7 @@ static void tag_project_function(SQLFunctionCtx *pCtx) {
assert(pCtx->inputBytes == pCtx->outputBytes);
for (int32_t i = 0; i < pCtx->size; ++i) {
char* output = pCtx->aOutputBuf;
if (pCtx->tag.nType == TSDB_DATA_TYPE_BINARY || pCtx->tag.nType == TSDB_DATA_TYPE_NCHAR) {
varDataSetLen(output, pCtx->tag.nLen);
tVariantDump(&pCtx->tag, varDataVal(output), pCtx->outputType);
} else {
tVariantDump(&pCtx->tag, output, pCtx->outputType);
}
tVariantDump(&pCtx->tag, pCtx->aOutputBuf, pCtx->outputType, true);
pCtx->aOutputBuf += pCtx->outputBytes;
}
......@@ -2997,13 +2990,7 @@ static void tag_project_function(SQLFunctionCtx *pCtx) {
static void tag_project_function_f(SQLFunctionCtx *pCtx, int32_t index) {
INC_INIT_VAL(pCtx, 1);
char* output = pCtx->aOutputBuf;
if (pCtx->tag.nType == TSDB_DATA_TYPE_BINARY || pCtx->tag.nType == TSDB_DATA_TYPE_NCHAR) {
*(int16_t*) output = pCtx->tag.nLen;
output += VARSTR_HEADER_SIZE;
}
tVariantDump(&pCtx->tag, output, pCtx->tag.nType);
tVariantDump(&pCtx->tag, pCtx->aOutputBuf, pCtx->tag.nType, true);
pCtx->aOutputBuf += pCtx->outputBytes;
}
......@@ -3016,30 +3003,12 @@ static void tag_project_function_f(SQLFunctionCtx *pCtx, int32_t index) {
*/
static void tag_function(SQLFunctionCtx *pCtx) {
SET_VAL(pCtx, 1, 1);
char* output = pCtx->aOutputBuf;
// todo refactor to dump length presented string(var string)
if (pCtx->tag.nType == TSDB_DATA_TYPE_BINARY || pCtx->tag.nType == TSDB_DATA_TYPE_NCHAR) {
*(int16_t*) output = pCtx->tag.nLen;
output += VARSTR_HEADER_SIZE;
}
tVariantDump(&pCtx->tag, output, pCtx->tag.nType);
tVariantDump(&pCtx->tag, pCtx->aOutputBuf, pCtx->tag.nType, true);
}
static void tag_function_f(SQLFunctionCtx *pCtx, int32_t index) {
SET_VAL(pCtx, 1, 1);
char* output = pCtx->aOutputBuf;
// todo refactor to dump length presented string(var string)
if (pCtx->tag.nType == TSDB_DATA_TYPE_BINARY || pCtx->tag.nType == TSDB_DATA_TYPE_NCHAR) {
*(int16_t*) output = pCtx->tag.nLen;
output += VARSTR_HEADER_SIZE;
}
tVariantDump(&pCtx->tag, output, pCtx->tag.nType);
tVariantDump(&pCtx->tag, pCtx->aOutputBuf, pCtx->tag.nType, true);
}
static void copy_function(SQLFunctionCtx *pCtx) {
......@@ -3853,15 +3822,15 @@ void twa_function_finalizer(SQLFunctionCtx *pCtx) {
}
/**
* param[1]: default value/previous value of specified timestamp
* param[2]: next value of specified timestamp
* param[3]: denotes if the result is a precious result or interpolation results
*
* @param pCtx
*/
static void interp_function(SQLFunctionCtx *pCtx) {
// at this point, the value is existed, return directly
if (pCtx->param[3].i64Key == 1) {
SResultInfo *pResInfo = GET_RES_INFO(pCtx);
SInterpInfoDetail* pInfo = pResInfo->interResultBuf;
if (pCtx->size == 1) {
char *pData = GET_INPUT_CHAR(pCtx);
assignVal(pCtx->aOutputBuf, pData, pCtx->inputBytes, pCtx->inputType);
} else {
......@@ -3869,68 +3838,57 @@ static void interp_function(SQLFunctionCtx *pCtx) {
* use interpolation to generate the result.
* Note: the result of primary timestamp column uses the timestamp specified by user in the query sql
*/
assert(pCtx->param[3].i64Key == 2);
SInterpInfo interpInfo = *(SInterpInfo *)pCtx->aOutputBuf;
SInterpInfoDetail *pInfoDetail = interpInfo.pInterpDetail;
assert(pCtx->size == 2);
if (pInfo->type == TSDB_FILL_NONE) { // set no output result
return;
}
/* set no output result */
if (pInfoDetail->type == TSDB_FILL_NONE) {
pCtx->param[3].i64Key = 0;
} else if (pInfoDetail->primaryCol == 1) {
*(TSKEY *)pCtx->aOutputBuf = pInfoDetail->ts;
if (pInfo->primaryCol == 1) {
*(TSKEY *) pCtx->aOutputBuf = pInfo->ts;
} else {
if (pInfoDetail->type == TSDB_FILL_NULL) {
if (pInfo->type == TSDB_FILL_NULL) {
if (pCtx->outputType == TSDB_DATA_TYPE_BINARY || pCtx->outputType == TSDB_DATA_TYPE_NCHAR) {
setVardataNull(pCtx->aOutputBuf, pCtx->outputType);
} else {
setNull(pCtx->aOutputBuf, pCtx->outputType, pCtx->outputBytes);
}
} else if (pInfoDetail->type == TSDB_FILL_SET_VALUE) {
tVariantDump(&pCtx->param[1], pCtx->aOutputBuf, pCtx->inputType);
} else if (pInfoDetail->type == TSDB_FILL_PREV) {
char *data = pCtx->param[1].pz;
char *pVal = data + TSDB_KEYSIZE;
if (pCtx->outputType == TSDB_DATA_TYPE_FLOAT) {
float v = GET_DOUBLE_VAL(pVal);
assignVal(pCtx->aOutputBuf, (const char*) &v, pCtx->outputBytes, pCtx->outputType);
} else {
assignVal(pCtx->aOutputBuf, pVal, pCtx->outputBytes, pCtx->outputType);
}
SET_VAL(pCtx, pCtx->size, 1);
} else if (pInfo->type == TSDB_FILL_SET_VALUE) {
tVariantDump(&pCtx->param[1], pCtx->aOutputBuf, pCtx->inputType, true);
} else if (pInfo->type == TSDB_FILL_PREV) {
char *data = GET_INPUT_CHAR_INDEX(pCtx, 0);
assignVal(pCtx->aOutputBuf, data, pCtx->outputBytes, pCtx->outputType);
} else if (pInfoDetail->type == TSDB_FILL_LINEAR) {
char *data1 = pCtx->param[1].pz;
char *data2 = pCtx->param[2].pz;
SET_VAL(pCtx, pCtx->size, 1);
} else if (pInfo->type == TSDB_FILL_LINEAR) {
char *data1 = GET_INPUT_CHAR_INDEX(pCtx, 0);
char *data2 = GET_INPUT_CHAR_INDEX(pCtx, 1);
char *pVal1 = data1 + TSDB_KEYSIZE;
char *pVal2 = data2 + TSDB_KEYSIZE;
TSKEY key1 = pCtx->ptsList[0];
TSKEY key2 = pCtx->ptsList[1];
SPoint point1 = {.key = *(TSKEY *)data1, .val = &pCtx->param[1].i64Key};
SPoint point2 = {.key = *(TSKEY *)data2, .val = &pCtx->param[2].i64Key};
SPoint point1 = {.key = key1, .val = data1};
SPoint point2 = {.key = key2, .val = data2};
SPoint point = {.key = pInfoDetail->ts, .val = pCtx->aOutputBuf};
SPoint point = {.key = pInfo->ts, .val = pCtx->aOutputBuf};
int32_t srcType = pCtx->inputType;
if ((srcType >= TSDB_DATA_TYPE_TINYINT && srcType <= TSDB_DATA_TYPE_BIGINT) ||
srcType == TSDB_DATA_TYPE_TIMESTAMP || srcType == TSDB_DATA_TYPE_DOUBLE) {
point1.val = pVal1;
point2.val = pVal2;
point1.val = data1;
point2.val = data2;
if (isNull(pVal1, srcType) || isNull(pVal2, srcType)) {
if (isNull(data1, srcType) || isNull(data2, srcType)) {
setNull(pCtx->aOutputBuf, srcType, pCtx->inputBytes);
} else {
taosDoLinearInterpolation(pCtx->outputType, &point1, &point2, &point);
}
} else if (srcType == TSDB_DATA_TYPE_FLOAT) {
float v1 = GET_DOUBLE_VAL(pVal1);
float v2 = GET_DOUBLE_VAL(pVal2);
point1.val = data1;
point2.val = data2;
point1.val = &v1;
point2.val = &v2;
if (isNull(pVal1, srcType) || isNull(pVal2, srcType)) {
if (isNull(data1, srcType) || isNull(data2, srcType)) {
setNull(pCtx->aOutputBuf, srcType, pCtx->inputBytes);
} else {
taosDoLinearInterpolation(pCtx->outputType, &point1, &point2, &point);
......@@ -3938,7 +3896,7 @@ static void interp_function(SQLFunctionCtx *pCtx) {
} else {
if (srcType == TSDB_DATA_TYPE_BINARY || srcType == TSDB_DATA_TYPE_NCHAR) {
setVardataNull(pCtx->aOutputBuf, pCtx->inputBytes);
setVardataNull(pCtx->aOutputBuf, pCtx->inputType);
} else {
setNull(pCtx->aOutputBuf, srcType, pCtx->inputBytes);
}
......@@ -3946,15 +3904,8 @@ static void interp_function(SQLFunctionCtx *pCtx) {
}
}
free(interpInfo.pInterpDetail);
}
pCtx->size = pCtx->param[3].i64Key;
tVariantDestroy(&pCtx->param[1]);
tVariantDestroy(&pCtx->param[2]);
// data in the check operation are all null, not output
SET_VAL(pCtx, pCtx->size, 1);
}
......@@ -4910,7 +4861,7 @@ SQLAggFuncElem aAggs[] = {{
"interp",
TSDB_FUNC_INTERP,
TSDB_FUNC_INTERP,
TSDB_FUNCSTATE_SO | TSDB_FUNCSTATE_OF | TSDB_FUNCSTATE_STABLE | TSDB_FUNCSTATE_NEED_TS,
TSDB_FUNCSTATE_SO | TSDB_FUNCSTATE_OF | TSDB_FUNCSTATE_STABLE | TSDB_FUNCSTATE_NEED_TS ,
function_setup,
interp_function,
do_sum_f, // todo filter handle
......@@ -4918,7 +4869,7 @@ SQLAggFuncElem aAggs[] = {{
doFinalizer,
noop1,
copy_function,
no_data_info,
data_req_load_info,
},
{
// 28
......
src/client/src/tscSQLParser.c
浏览文件 @ 3f64b14a
......@@ -142,7 +142,7 @@ static int setColumnFilterInfoForTimestamp(SQueryInfo* pQueryInfo, tVariant* pVa
return invalidSqlErrMsg(pQueryInfo->msg, msg);
}
} else {
if (tVariantDump(pVar, (char*)&time, TSDB_DATA_TYPE_BIGINT)) {
if (tVariantDump(pVar, (char*)&time, TSDB_DATA_TYPE_BIGINT, true)) {
return invalidSqlErrMsg(pQueryInfo->msg, msg);
}
}
......@@ -1403,7 +1403,6 @@ int32_t addProjectionExprAndResultField(SQueryInfo* pQueryInfo, tSQLExprItem* pI
SSchema colSchema = {.type = TSDB_DATA_TYPE_BINARY, .bytes = TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE};
strcpy(colSchema.name, TSQL_TBNAME_L);
pQueryInfo->type = TSDB_QUERY_TYPE_STABLE_QUERY;
tscAddSpecialColumnForSelect(pQueryInfo, startPos, TSDB_FUNC_TAGPRJ, &index, &colSchema, true);
} else {
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, index.tableIndex);
......@@ -1595,7 +1594,7 @@ int32_t addExprAndResultField(SQueryInfo* pQueryInfo, int32_t colIndex, tSQLExpr
int16_t resultType = 0;
int16_t resultSize = 0;
int16_t intermediateResSize = 0;
int32_t intermediateResSize = 0;
int16_t functionID = 0;
if (changeFunctionID(optr, &functionID) != TSDB_CODE_SUCCESS) {
......@@ -1628,14 +1627,14 @@ int32_t addExprAndResultField(SQueryInfo* pQueryInfo, int32_t colIndex, tSQLExpr
if (optr == TK_LEASTSQUARES) {
/* set the leastsquares parameters */
char val[8] = {0};
if (tVariantDump(&pParamElem[1].pNode->val, val, TSDB_DATA_TYPE_DOUBLE) < 0) {
if (tVariantDump(&pParamElem[1].pNode->val, val, TSDB_DATA_TYPE_DOUBLE, true) < 0) {
return TSDB_CODE_INVALID_SQL;
}
addExprParams(pExpr, val, TSDB_DATA_TYPE_DOUBLE, DOUBLE_BYTES, 0);
memset(val, 0, tListLen(val));
if (tVariantDump(&pParamElem[2].pNode->val, val, TSDB_DATA_TYPE_DOUBLE) < 0) {
if (tVariantDump(&pParamElem[2].pNode->val, val, TSDB_DATA_TYPE_DOUBLE, true) < 0) {
return TSDB_CODE_INVALID_SQL;
}
......@@ -1795,7 +1794,7 @@ int32_t addExprAndResultField(SQueryInfo* pQueryInfo, int32_t colIndex, tSQLExpr
SSqlExpr* pExpr = NULL;
if (optr == TK_PERCENTILE || optr == TK_APERCENTILE) {
tVariantDump(pVariant, val, TSDB_DATA_TYPE_DOUBLE);
tVariantDump(pVariant, val, TSDB_DATA_TYPE_DOUBLE, true);
double dp = GET_DOUBLE_VAL(val);
if (dp < 0 || dp > TOP_BOTTOM_QUERY_LIMIT) {
......@@ -1818,7 +1817,7 @@ int32_t addExprAndResultField(SQueryInfo* pQueryInfo, int32_t colIndex, tSQLExpr
pExpr = tscSqlExprAppend(pQueryInfo, functionId, &index, resultType, resultSize, resultSize, false);
addExprParams(pExpr, val, TSDB_DATA_TYPE_DOUBLE, sizeof(double), 0);
} else {
tVariantDump(pVariant, val, TSDB_DATA_TYPE_BIGINT);
tVariantDump(pVariant, val, TSDB_DATA_TYPE_BIGINT, true);
int64_t nTop = *((int32_t*)val);
if (nTop <= 0 || nTop > 100) { // todo use macro
......@@ -1902,7 +1901,7 @@ int32_t addExprAndResultField(SQueryInfo* pQueryInfo, int32_t colIndex, tSQLExpr
int16_t bytes = 0;
int16_t type = 0;
int16_t inter = 0;
int32_t inter = 0;
int32_t ret = getResultDataInfo(s.type, s.bytes, TSDB_FUNC_TID_TAG, 0, &type, &bytes, &inter, 0, 0);
assert(ret == TSDB_CODE_SUCCESS);
......@@ -2288,7 +2287,7 @@ int32_t tscTansformSQLFuncForSTableQuery(SQueryInfo* pQueryInfo) {
int16_t bytes = 0;
int16_t type = 0;
int16_t intermediateBytes = 0;
int32_t interBytes = 0;
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t k = 0; k < size; ++k) {
......@@ -2302,13 +2301,13 @@ int32_t tscTansformSQLFuncForSTableQuery(SQueryInfo* pQueryInfo) {
(functionId >= TSDB_FUNC_FIRST_DST && functionId <= TSDB_FUNC_LAST_DST) ||
(functionId >= TSDB_FUNC_RATE && functionId <= TSDB_FUNC_AVG_IRATE)) {
if (getResultDataInfo(pSrcSchema->type, pSrcSchema->bytes, functionId, pExpr->param[0].i64Key, &type, &bytes,
&intermediateBytes, 0, true) != TSDB_CODE_SUCCESS) {
&interBytes, 0, true) != TSDB_CODE_SUCCESS) {
return TSDB_CODE_INVALID_SQL;
}
tscSqlExprUpdate(pQueryInfo, k, functionId, pExpr->colInfo.colIndex, TSDB_DATA_TYPE_BINARY, bytes);
// todo refactor
pExpr->interBytes = intermediateBytes;
pExpr->interBytes = interBytes;
}
}
......@@ -2328,12 +2327,9 @@ void tscRestoreSQLFuncForSTableQuery(SQueryInfo* pQueryInfo) {
SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, i);
SSchema* pSchema = tscGetTableColumnSchema(pTableMetaInfo->pTableMeta, pExpr->colInfo.colIndex);
// if (/*(pExpr->functionId >= TSDB_FUNC_FIRST_DST && pExpr->functionId <= TSDB_FUNC_LAST_DST) ||
// (pExpr->functionId >= TSDB_FUNC_SUM && pExpr->functionId <= TSDB_FUNC_MAX) ||
// pExpr->functionId == TSDB_FUNC_LAST_ROW*/) {
// the final result size and type in the same as query on single table.
// so here, set the flag to be false;
int16_t inter = 0;
int32_t inter = 0;
int32_t functionId = pExpr->functionId;
if (functionId >= TSDB_FUNC_TS && functionId <= TSDB_FUNC_DIFF) {
......@@ -2348,7 +2344,6 @@ void tscRestoreSQLFuncForSTableQuery(SQueryInfo* pQueryInfo) {
getResultDataInfo(pSchema->type, pSchema->bytes, functionId, 0, &pExpr->resType, &pExpr->resBytes,
&inter, 0, false);
// }
}
}
......@@ -2631,23 +2626,23 @@ static int32_t doExtractColumnFilterInfo(SQueryInfo* pQueryInfo, SColumnFilterIn
}
if (pExpr->nSQLOptr == TK_LE || pExpr->nSQLOptr == TK_LT) {
tVariantDump(&pRight->val, (char*)&pColumnFilter->upperBndd, colType);
tVariantDump(&pRight->val, (char*)&pColumnFilter->upperBndd, colType, false);
} else { // TK_GT,TK_GE,TK_EQ,TK_NE are based on the pColumn->lowerBndd
if (colType == TSDB_DATA_TYPE_BINARY) {
pColumnFilter->pz = (int64_t)calloc(1, pRight->val.nLen + 1);
pColumnFilter->len = pRight->val.nLen;
tVariantDump(&pRight->val, (char*)pColumnFilter->pz, colType);
tVariantDump(&pRight->val, (char*)pColumnFilter->pz, colType, false);
} else if (colType == TSDB_DATA_TYPE_NCHAR) {
// pRight->val.nLen + 1 is larger than the actual nchar string length
pColumnFilter->pz = (int64_t)calloc(1, (pRight->val.nLen + 1) * TSDB_NCHAR_SIZE);
tVariantDump(&pRight->val, (char*)pColumnFilter->pz, colType);
tVariantDump(&pRight->val, (char*)pColumnFilter->pz, colType, false);
size_t len = wcslen((wchar_t*)pColumnFilter->pz);
pColumnFilter->len = len * TSDB_NCHAR_SIZE;
} else {
tVariantDump(&pRight->val, (char*)&pColumnFilter->lowerBndd, colType);
tVariantDump(&pRight->val, (char*)&pColumnFilter->lowerBndd, colType, false);
}
}
......@@ -3336,9 +3331,8 @@ static int32_t handleExprInQueryCond(SQueryInfo* pQueryInfo, tSQLExpr** pExpr, S
*pExpr = NULL; // remove this expression
*type = TSQL_EXPR_TS;
} else if (index.columnIndex >= tscGetNumOfColumns(pTableMeta) ||
index.columnIndex == TSDB_TBNAME_COLUMN_INDEX) { // query on tags
// check for tag query condition
} else if (index.columnIndex >= tscGetNumOfColumns(pTableMeta) || index.columnIndex == TSDB_TBNAME_COLUMN_INDEX) {
// query on tags, check for tag query condition
if (UTIL_TABLE_IS_NORMAL_TABLE(pTableMetaInfo)) {
return invalidSqlErrMsg(pQueryInfo->msg, msg1);
}
......@@ -3933,7 +3927,7 @@ int32_t getTimeRange(STimeWindow* win, tSQLExpr* pRight, int32_t optr, int16_t t
* failed to parse timestamp in regular formation, try next
* it may be a epoch time in string format
*/
tVariantDump(&pRight->val, (char*)&val, TSDB_DATA_TYPE_BIGINT);
tVariantDump(&pRight->val, (char*)&val, TSDB_DATA_TYPE_BIGINT, true);
/*
* transfer it into MICROSECOND format if it is a string, since for
......@@ -4070,14 +4064,13 @@ int32_t parseFillClause(SQueryInfo* pQueryInfo, SQuerySQL* pQuerySQL) {
continue;
}
int32_t ret = tVariantDump(&pFillToken->a[j].pVar, (char*)&pQueryInfo->fillVal[i], pFields->type);
int32_t ret = tVariantDump(&pFillToken->a[j].pVar, (char*)&pQueryInfo->fillVal[i], pFields->type, true);
if (ret != TSDB_CODE_SUCCESS) {
return invalidSqlErrMsg(pQueryInfo->msg, msg);
}
}
if ((pFillToken->nExpr < size) ||
((pFillToken->nExpr - 1 < size) && (tscIsPointInterpQuery(pQueryInfo)))) {
if ((pFillToken->nExpr < size) || ((pFillToken->nExpr - 1 < size) && (tscIsPointInterpQuery(pQueryInfo)))) {
tVariantListItem* lastItem = &pFillToken->a[pFillToken->nExpr - 1];
for (int32_t i = numOfFillVal; i < size; ++i) {
......@@ -4086,7 +4079,7 @@ int32_t parseFillClause(SQueryInfo* pQueryInfo, SQuerySQL* pQuerySQL) {
if (pFields->type == TSDB_DATA_TYPE_BINARY || pFields->type == TSDB_DATA_TYPE_NCHAR) {
setVardataNull((char*) &pQueryInfo->fillVal[i], pFields->type);
} else {
tVariantDump(&lastItem->pVar, (char*)&pQueryInfo->fillVal[i], pFields->type);
tVariantDump(&lastItem->pVar, (char*)&pQueryInfo->fillVal[i], pFields->type, true);
}
}
}
......@@ -4168,6 +4161,10 @@ int32_t parseOrderbyClause(SQueryInfo* pQueryInfo, SQuerySQL* pQuerySql, SSchema
if (index.columnIndex >= tscGetNumOfColumns(pTableMetaInfo->pTableMeta)) {
int32_t relTagIndex = index.columnIndex - tscGetNumOfColumns(pTableMetaInfo->pTableMeta);
// it is a tag column
if (pQueryInfo->groupbyExpr.columnInfo == NULL) {
return invalidSqlErrMsg(pQueryInfo->msg, msg2);
}
SColIndex* pColIndex = taosArrayGet(pQueryInfo->groupbyExpr.columnInfo, 0);
if (relTagIndex == pColIndex->colIndex) {
orderByTags = true;
......@@ -4420,10 +4417,10 @@ int32_t setAlterTableInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
}
SSchema* pTagsSchema = tscGetTableColumnSchema(pTableMetaInfo->pTableMeta, columnIndex.columnIndex);
if (tVariantDump(&pVarList->a[1].pVar, pAlterSQL->tagData.data /*pCmd->payload*/, pTagsSchema->type) !=
TSDB_CODE_SUCCESS) {
if (tVariantDump(&pVarList->a[1].pVar, pAlterSQL->tagData.data, pTagsSchema->type, true) != TSDB_CODE_SUCCESS) {
return invalidSqlErrMsg(pQueryInfo->msg, msg13);
}
pAlterSQL->tagData.dataLen = pTagsSchema->bytes;
// validate the length of binary
......@@ -4680,7 +4677,7 @@ int32_t parseLimitClause(SQueryInfo* pQueryInfo, int32_t clauseIndex, SQuerySQL*
const char* msg0 = "soffset/offset can not be less than 0";
const char* msg1 = "slimit/soffset only available for STable query";
const char* msg2 = "function not supported on table";
const char* msg2 = "functions mixed up in table query";
const char* msg3 = "slimit/soffset can not apply to projection query";
// handle the limit offset value, validate the limit
......@@ -4764,12 +4761,20 @@ int32_t parseLimitClause(SQueryInfo* pQueryInfo, int32_t clauseIndex, SQuerySQL*
size_t size = taosArrayGetSize(pQueryInfo->exprList);
bool hasTags = false;
bool hasOtherFunc = false;
// filter the query functions operating on "tbname" column that are not supported by normal columns.
for (int32_t i = 0; i < size; ++i) {
SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, i);
if (pExpr->colInfo.colIndex == TSDB_TBNAME_COLUMN_INDEX) {
return invalidSqlErrMsg(pQueryInfo->msg, msg2);
if (TSDB_COL_IS_TAG(pExpr->colInfo.flag)) {
hasTags = true;
} else {
hasOtherFunc = true;
}
}
if (hasTags && hasOtherFunc) {
return invalidSqlErrMsg(pQueryInfo->msg, msg2);
}
}
......@@ -5571,21 +5576,9 @@ int32_t doCheckForCreateFromStable(SSqlObj* pSql, SSqlInfo* pInfo) {
if (pList->a[i].pVar.nLen + VARSTR_HEADER_SIZE > pTagSchema[i].bytes) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg3);
}
ret = tVariantDump(&(pList->a[i].pVar), varDataVal(tagVal), pTagSchema[i].type);
if (pList->a[i].pVar.nType == TSDB_DATA_TYPE_NULL) {
if (pTagSchema[i].type == TSDB_DATA_TYPE_BINARY) {
varDataSetLen(tagVal, sizeof(uint8_t));
} else {
varDataSetLen(tagVal, sizeof(uint32_t));
}
} else { // todo refactor
varDataSetLen(tagVal, pList->a[i].pVar.nLen);
}
} else {
ret = tVariantDump(&(pList->a[i].pVar), tagVal, pTagSchema[i].type);
}
ret = tVariantDump(&(pList->a[i].pVar), tagVal, pTagSchema[i].type, true);
if (ret != TSDB_CODE_SUCCESS) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg4);
}
......@@ -5845,7 +5838,7 @@ int32_t doCheckForQuery(SSqlObj* pSql, SQuerySQL* pQuerySql, int32_t index) {
pQueryInfo->window.ekey = pQueryInfo->window.ekey / 1000;
}
} else { // set the time rang
pQueryInfo->window.skey = 0;
pQueryInfo->window.skey = TSKEY_INITIAL_VAL;
pQueryInfo->window.ekey = INT64_MAX;
}
......
src/client/src/tscSecondaryMerge.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/client/src/tscServer.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/client/src/tscSql.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/client/src/tscSubquery.c
浏览文件 @ 3f64b14a
......@@ -1084,7 +1084,7 @@ int32_t tscLaunchJoinSubquery(SSqlObj *pSql, int16_t tableIndex, SJoinSupporter
int16_t bytes = 0;
int16_t type = 0;
int16_t inter = 0;
int32_t inter = 0;
getResultDataInfo(s.type, s.bytes, TSDB_FUNC_TID_TAG, 0, &type, &bytes, &inter, 0, 0);
......@@ -1770,6 +1770,8 @@ int32_t tscHandleMultivnodeInsert(SSqlObj *pSql) {
*/
pNew->fetchFp = pNew->fp;
pSql->pSubs[i] = pNew;
pNew->fetchFp = pNew->fp;
tscTrace("%p sub:%p create subObj success. orderOfSub:%d", pSql, pNew, i);
}
......
src/client/src/tscUtil.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/common/inc/tglobal.h
浏览文件 @ 3f64b14a
......@@ -94,6 +94,10 @@ extern int32_t tsMaxTables;
extern char tsDefaultDB[];
extern char tsDefaultUser[];
extern char tsDefaultPass[];
extern char tsMqttBrokerAddress[];
extern char tsMqttBrokerClientId[];
extern int32_t tsMaxConnections;
extern int32_t tsBalanceInterval;
......
src/common/src/tdataformat.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/common/src/tglobal.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/common/src/ttypes.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/dnode/src/dnodeModule.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/inc/taosdef.h
浏览文件 @ 3f64b14a
此差异已折叠。
src/inc/taosmsg.h
浏览文件 @ 3f64b14a
此差异已折叠。
src/inc/mqtt.h → src/inc/tmqtt.h
浏览文件 @ 3f64b14a
此差异已折叠。
src/inc/tsdb.h
浏览文件 @ 3f64b14a
此差异已折叠。
src/mnode/src/mnodeDb.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/plugins/mqtt/CMakeLists.txt
浏览文件 @ 3f64b14a
此差异已折叠。
src/plugins/mqtt/inc/mqttInit.h 0 → 100644
浏览文件 @ 3f64b14a
此差异已折叠。
src/plugins/mqtt/inc/mqttPayload.h 0 → 100644
浏览文件 @ 3f64b14a
此差异已折叠。
src/plugins/mqtt/inc/mqttSystem.h
浏览文件 @ 3f64b14a
此差异已折叠。
src/plugins/mqtt/src/mqttPayload.c 0 → 100644
浏览文件 @ 3f64b14a
此差异已折叠。
src/plugins/mqtt/src/mqttSystem.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/query/inc/qExecutor.h
浏览文件 @ 3f64b14a
此差异已折叠。
src/query/inc/qextbuffer.h
浏览文件 @ 3f64b14a
此差异已折叠。
src/query/inc/qfill.h
浏览文件 @ 3f64b14a
此差异已折叠。
src/query/inc/qresultBuf.h
浏览文件 @ 3f64b14a
此差异已折叠。
src/query/inc/tsqlfunction.h
浏览文件 @ 3f64b14a
此差异已折叠。
src/query/inc/tvariant.h
浏览文件 @ 3f64b14a
此差异已折叠。
src/query/src/qExecutor.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/query/src/qFilterFunc.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/query/src/qUtil.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/query/src/qfill.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/query/src/qresultBuf.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/query/src/tvariant.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/rpc/src/rpcMain.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/tsdb/src/tsdbFile.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/tsdb/src/tsdbMeta.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/tsdb/src/tsdbRWHelper.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/tsdb/src/tsdbRead.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/util/inc/hash.h
浏览文件 @ 3f64b14a
此差异已折叠。
src/util/inc/tutil.h
浏览文件 @ 3f64b14a
此差异已折叠。
src/util/src/hash.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/util/src/tcompare.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/util/src/ttimer.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/util/src/tutil.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/util/tests/hashTest.cpp
浏览文件 @ 3f64b14a
此差异已折叠。
src/vnode/src/vnodeMain.c
浏览文件 @ 3f64b14a
此差异已折叠。
src/vnode/src/vnodeWrite.c
浏览文件 @ 3f64b14a
此差异已折叠。
tests/examples/c/demo.c
浏览文件 @ 3f64b14a
此差异已折叠。
tests/pytest/fulltest.sh
浏览文件 @ 3f64b14a
此差异已折叠。
tests/pytest/smoketest.sh
浏览文件 @ 3f64b14a
此差异已折叠。
tests/pytest/util/dnodes.py
浏览文件 @ 3f64b14a
此差异已折叠。
tests/script/jenkins/basic.txt
浏览文件 @ 3f64b14a
此差异已折叠。
tests/script/jenkins/sync.txt
浏览文件 @ 3f64b14a
此差异已折叠。
tests/script/sh/cfg.sh
浏览文件 @ 3f64b14a
此差异已折叠。
tests/script/sh/clear.sh
浏览文件 @ 3f64b14a
此差异已折叠。
tests/script/sh/deploy.sh
浏览文件 @ 3f64b14a
此差异已折叠。
tests/script/sh/exec.sh
浏览文件 @ 3f64b14a
此差异已折叠。
tests/script/sh/exec_tarbitrator.sh
浏览文件 @ 3f64b14a
此差异已折叠。
tests/script/sh/exec_up.sh 已删除 100755 → 0
浏览文件 @ a6e2ef7b
此差异已折叠。
tests/script/test.sh
浏览文件 @ 3f64b14a
此差异已折叠。
tests/script/unique/big/balance.sim
浏览文件 @ 3f64b14a
此差异已折叠。
tests/script/unique/big/tcp.sim
浏览文件 @ 3f64b14a
此差异已折叠。
tests/script/unique/db/commit.sim
浏览文件 @ 3f64b14a
此差异已折叠。
tests/script/unique/db/delete.sim
浏览文件 @ 3f64b14a
此差异已折叠。
tests/script/unique/http/admin.sim
浏览文件 @ 3f64b14a
此差异已折叠。
tests/script/unique/vnode/many.sim
浏览文件 @ 3f64b14a
此差异已折叠。
tests/test-all.sh
浏览文件 @ 3f64b14a
此差异已折叠。
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或