MQTT_DEMO
- 1
- 29
- 30
- 31 #include <errno.h>
- 32 #include <fcntl.h>
- 33 #include <stdio.h>
- 34 #include <stdlib.h>
- 35 #include <string.h>
- 36 #ifndef WIN32
- 37 #include <unistd.h>
- 38 #else
- 39 #include <process.h>
- 40 #include <winsock2.h>
- 41 #define snprintf sprintf_s
- 42 #endif
- 43
- 44 #include <mosquitto.h>
- 45
- 46 #define MSGMODE_NONE 0
- 47 #define MSGMODE_CMD 1
- 48 #define MSGMODE_STDIN_LINE 2
- 49 #define MSGMODE_STDIN_FILE 3
- 50 #define MSGMODE_FILE 4
- 51 #define MSGMODE_NULL 5
- 52
- 53 #define STATUS_CONNECTING 0
- 54 #define STATUS_CONNACK_RECVD 1
- 55
- 56 static char *topic = NULL;
- 57 static char *message = NULL;
- 58 static long msglen = 0;
- 59 static int qos = 0;
- 60 static int retain = 0;
- 61 static int mode = MSGMODE_NONE;
- 62 static int status = STATUS_CONNECTING;
- 63 static uint16_t mid_sent = 0;
- 64 static bool connected = true;
- 65 static char *username = NULL;
- 66 static char *password = NULL;
- 67 static bool disconnect_sent = false;
- 68 static bool quiet = false;
- 69
- 70 void my_connect_callback(void *obj, int result)
- 71 {
- 72
- 73 struct mosquitto *mosq = obj;
- 74 int rc = MOSQ_ERR_SUCCESS;
- 75
- 76 if(!result){
- 77 switch(mode){
- 78 case MSGMODE_CMD:
- 79 case MSGMODE_FILE:
- 80 case MSGMODE_STDIN_FILE:
- 81 rc = mosquitto_publish(mosq, &mid_sent, topic, msglen, (uint8_t *)message, qos, retain);
- 82 break;
- 83 case MSGMODE_NULL:
- 84 rc = mosquitto_publish(mosq, &mid_sent, topic, 0, NULL, qos, retain);
- 85 break;
- 86 case MSGMODE_STDIN_LINE:
- 87 status = STATUS_CONNACK_RECVD;
- 88 break;
- 89 }
- 90 if(rc){
- 91 if(!quiet){
- 92 switch(rc){
- 93 case MOSQ_ERR_INVAL:
- 94 fprintf(stderr, "Error: Invalid input. Does your topic contain '+' or '#'?
");
- 95 break;
- 96 case MOSQ_ERR_NOMEM:
- 97 fprintf(stderr, "Error: Out of memory when trying to publish message.
");
- 98 break;
- 99 case MOSQ_ERR_NO_CONN:
- 100 fprintf(stderr, "Error: Client not connected when trying to publish.
");
- 101 break;
- 102 case MOSQ_ERR_PROTOCOL:
- 103 fprintf(stderr, "Error: Protocol error when communicating with broker.
");
- 104 break;
- 105 case MOSQ_ERR_PAYLOAD_SIZE:
- 106 fprintf(stderr, "Error: Message payload is too large.
");
- 107 break;
- 108 }
- 109 }
- 110 mosquitto_disconnect(mosq);
- 111 }
- 112 }else{
- 113 switch(result){
- 114 case 1:
- 115 if(!quiet) fprintf(stderr, "Connection Refused: unacceptable protocol version
");
- 116 break;
- 117 case 2:
- 118 if(!quiet) fprintf(stderr, "Connection Refused: identifier rejected
");
- 119 break;
- 120 case 3:
- 121 if(!quiet) fprintf(stderr, "Connection Refused: broker unavailable
");
- 122 break;
- 123 case 4:
- 124 if(!quiet) fprintf(stderr, "Connection Refused: bad user name or password
");
- 125 break;
- 126 case 5:
- 127 if(!quiet) fprintf(stderr, "Connection Refused: not authorised
");
- 128 break;
- 129 default:
- 130 if(!quiet) fprintf(stderr, "Connection Refused: unknown reason
");
- 131 break;
- 132 }
- 133 }
- 134 }
- 135
- 136 void my_disconnect_callback(void *obj)
- 137 {
- 138
- 139 connected = false;
- 140 }
- 141
- 142 void my_publish_callback(void *obj, uint16_t mid)
- 143 {
- 144
- 145 struct mosquitto *mosq = obj;
- 146
- 147 if(mode != MSGMODE_STDIN_LINE && disconnect_sent == false){
- 148 mosquitto_disconnect(mosq);
- 149 disconnect_sent = true;
- 150 }
- 151 }
- 152
- 153 int load_stdin(void)
- 154 {
- 155
- 156 long pos = 0, rlen;
- 157 char buf[1024];
- 158
- 159 mode = MSGMODE_STDIN_FILE;
- 160
- 161 while(!feof(stdin)){
- 162 rlen = fread(buf, 1, 1024, stdin);
- 163 message = realloc(message, pos+rlen);
- 164 if(!message){
- 165 if(!quiet) fprintf(stderr, "Error: Out of memory.
");
- 166 return 1;
- 167 }
- 168 memcpy(&(message[pos]), buf, rlen);
- 169 pos += rlen;
- 170 }
- 171 msglen = pos;
- 172
- 173 if(!msglen){
- 174 if(!quiet) fprintf(stderr, "Error: Zero length input.
");
- 175 return 1;
- 176 }
- 177
- 178 return 0;
- 179 }
- 180
- 181 int load_file(const char *filename)
- 182 {
- 183 long pos, rlen;
- 184 FILE *fptr = NULL;
- 185
- 186 fptr = fopen(filename, "rb");
- 187 if(!fptr){
- 188 if(!quiet) fprintf(stderr, "Error: Unable to open file "%s".
", filename);
- 189 return 1;
- 190 }
- 191 mode = MSGMODE_FILE;
- 192 fseek(fptr, 0, SEEK_END);
- 193 msglen = ftell(fptr);
- 194 if(msglen > 268435455){
- 195 fclose(fptr);
- 196 if(!quiet) fprintf(stderr, "Error: File "%s" is too large (>268,435,455 bytes).
", filename);
- 197 return 1;
- 198 }
- 199 if(msglen == 0){
- 200 fclose(fptr);
- 201 if(!quiet) fprintf(stderr, "Error: File "%s" is empty.
", filename);
- 202 return 1;
- 203 }
- 204 fseek(fptr, 0, SEEK_SET);
- 205 message = malloc(msglen);
- 206 if(!message){
- 207 fclose(fptr);
- 208 if(!quiet) fprintf(stderr, "Error: Out of memory.
");
- 209 return 1;
- 210 }
- 211 pos = 0;
- 212
- 213 while(pos < msglen){
- 214 rlen = fread(&(message[pos]), sizeof(char), msglen-pos, fptr);
- 215 pos += rlen;
- 216 }
- 217 fclose(fptr);
- 218 return 0;
- 219 }
- 220
- 221 void print_usage(void)
- 222 {
- 223 printf("mosquitto_pub is a simple mqtt client that will publish a message on a single topic and exit.
");
- 224 printf("Usage: mosquitto_pub [-h host] [-p port] [-q qos] [-r] {-f file | -l | -n | -m message} -t topic
");
- 225 printf(" [-i id] [-I id_prefix]
");
- 226 printf(" [-d] [--quiet]
");
- 227 printf(" [-u username [-P password]]
");
- 228 printf(" [--will-topic [--will-payload payload] [--will-qos qos] [--will-retain]]
");
- 229 printf(" -d : enable debug messages.
");
- 230 printf(" -f : send the contents of a file as the message.
");
- 231 printf(" -h : mqtt host to connect to. Defaults to localhost.
");
- 232 printf(" -i : id to use for this client. Defaults to mosquitto_pub_ appended with the process id.
");
- 233 printf(" -I : define the client id as id_prefix appended with the process id. Useful for when the
");
- 234 printf(" broker is using the clientid_prefixes option.
");
- 235 printf(" -l : read messages from stdin, sending a separate message for each line.
");
- 236 printf(" -m : message payload to send.
");
- 237 printf(" -n : send a null (zero length) message.
");
- 238 printf(" -p : network port to connect to. Defaults to 1883.
");
- 239 printf(" -q : quality of service level to use for all messages. Defaults to 0.
");
- 240 printf(" -r : message should be retained.
");
- 241 printf(" -s : read message from stdin, sending the entire input as a message.
");
- 242 printf(" -t : mqtt topic to publish to.
");
- 243 printf(" -u : provide a username (requires MQTT 3.1 broker)
");
- 244 printf(" -P : provide a password (requires MQTT 3.1 broker)
");
- 245 printf(" --quiet : don't print error messages.
");
- 246 printf(" --will-payload : payload for the client Will, which is sent by the broker in case of
");
- 247 printf(" unexpected disconnection. If not given and will-topic is set, a zero
");
- 248 printf(" length message will be sent.
");
- 249 printf(" --will-qos : QoS level for the client Will.
");
- 250 printf(" --will-retain : if given, make the client Will retained.
");
- 251 printf(" --will-topic : the topic on which to publish the client Will.
");
- 252 printf("
See http://mosquitto.org/ for more information.
");
- 253 }
- 254
- 255 int main(int argc, char *argv[])
- 256 {
- 257 char *id = NULL;
- 258 char *id_prefix = NULL;
- 259 int i;
- 260 char *host = "localhost";
- 261 int port = 1883;
- 262 int keepalive = 60;
- 263 int opt;
- 264 char buf[1024];
- 265 bool debug = false;
- 266 struct mosquitto *mosq = NULL;
- 267 int rc;
- 268 int rc2;
- 269 char hostname[21];
- 270 char err[1024];
- 271
- 272 uint8_t *will_payload = NULL;
- 273 long will_payloadlen = 0;
- 274 int will_qos = 0;
- 275 bool will_retain = false;
- 276 char *will_topic = NULL;
- 277
- 278
- 279 for(i=1; i<argc; i++){
- 280 if(!strcmp(argv[i], "-p") || !strcmp(argv[i], "--port")){
- 281 if(i==argc-1){
- 282 fprintf(stderr, "Error: -p argument given but no port specified.
");
- 283 print_usage();
- 284 return 1;
- 285 }else{
- 286 port = atoi(argv[i+1]);
- 287 if(port<1 || port>65535){
- 288 fprintf(stderr, "Error: Invalid port given: %d
", port);
- 289 print_usage();
- 290 return 1;
- 291 }
- 292 }
- 293 i++;
- 294 }else if(!strcmp(argv[i], "-d") || !strcmp(argv[i], "--debug")){
- 295 debug = true;
- 296 }else if(!strcmp(argv[i], "-f") || !strcmp(argv[i], "--file")){
- 297 if(mode != MSGMODE_NONE){
- 298 fprintf(stderr, "Error: Only one type of message can be sent at once.
");
- 299 print_usage();
- 300 return 1;
- 301 }else if(i==argc-1){
- 302 fprintf(stderr, "Error: -f argument given but no file specified.
");
- 303 print_usage();
- 304 return 1;
- 305 }else{
- 306 if(load_file(argv[i+1])) return 1;
- 307 }
- 308 i++;
- 309 }else if(!strcmp(argv[i], "-h") || !strcmp(argv[i], "--host")){
- 310 if(i==argc-1){
- 311 fprintf(stderr, "Error: -h argument given but no host specified.
");
- 312 print_usage();
- 313 return 1;
- 314 }else{
- 315 host = argv[i+1];
- 316 }
- 317 i++;
- 318 }else if(!strcmp(argv[i], "-i") || !strcmp(argv[i], "--id")){
- 319 if(id_prefix){
- 320 fprintf(stderr, "Error: -i and -I argument cannot be used together.
");
- 321 print_usage();
- 322 return 1;
- 323 }
- 324 if(i==argc-1){
- 325 fprintf(stderr, "Error: -i argument given but no id specified.
");
- 326 print_usage();
- 327 return 1;
- 328 }else{
- 329 id = argv[i+1];
- 330 }
- 331 i++;
- 332 }else if(!strcmp(argv[i], "-I") || !strcmp(argv[i], "--id-prefix")){
- 333 if(id){
- 334 fprintf(stderr, "Error: -i and -I argument cannot be used together.
");
- 335 print_usage();
- 336 return 1;
- 337 }
- 338 if(i==argc-1){
- 339 fprintf(stderr, "Error: -I argument given but no id prefix specified.
");
- 340 print_usage();
- 341 return 1;
- 342 }else{
- 343 id_prefix = argv[i+1];
- 344 }
- 345 i++;
- 346 }else if(!strcmp(argv[i], "-l") || !strcmp(argv[i], "--stdin-line")){
- 347 if(mode != MSGMODE_NONE){
- 348 fprintf(stderr, "Error: Only one type of message can be sent at once.
");
- 349 print_usage();
- 350 return 1;
- 351 }else{
- 352 mode = MSGMODE_STDIN_LINE;
- 353 #ifndef WIN32
- 354 opt = fcntl(fileno(stdin), F_GETFL, 0);
- 355 if(opt == -1 || fcntl(fileno(stdin), F_SETFL, opt | O_NONBLOCK) == -1){
- 356 fprintf(stderr, "Error: Unable to set stdin to non-blocking.
");
- 357 return 1;
- 358 }
- 359 #endif
- 360 }
- 361 }else if(!strcmp(argv[i], "-m") || !strcmp(argv[i], "--message")){
- 362 if(mode != MSGMODE_NONE){
- 363 fprintf(stderr, "Error: Only one type of message can be sent at once.
");
- 364 print_usage();
- 365 return 1;
- 366 }else if(i==argc-1){
- 367 fprintf(stderr, "Error: -m argument given but no message specified.
");
- 368 print_usage();
- 369 return 1;
- 370 }else{
- 371 message = argv[i+1];
- 372 msglen = strlen(message);
- 373 mode = MSGMODE_CMD;
- 374 }
- 375 i++;
- 376 }else if(!strcmp(argv[i], "-n") || !strcmp(argv[i], "--null-message")){
- 377 if(mode != MSGMODE_NONE){
- 378 fprintf(stderr, "Error: Only one type of message can be sent at once.
");
- 379 print_usage();
- 380 return 1;
- 381 }else{
- 382 mode = MSGMODE_NULL;
- 383 }
- 384 }else if(!strcmp(argv[i], "-q") || !strcmp(argv[i], "--qos")){
- 385 if(i==argc-1){
- 386 fprintf(stderr, "Error: -q argument given but no QoS specified.
");
- 387 print_usage();
- 388 return 1;
- 389 }else{
- 390 qos = atoi(argv[i+1]);
- 391 if(qos<0 || qos>2){
- 392 fprintf(stderr, "Error: Invalid QoS given: %d
", qos);
- 393 print_usage();
- 394 return 1;
- 395 }
- 396 }
- 397 i++;
- 398 }else if(!strcmp(argv[i], "--quiet")){
- 399 quiet = true;
- 400 }else if(!strcmp(argv[i], "-r") || !strcmp(argv[i], "--retain")){
- 401 retain = 1;
- 402 }else if(!strcmp(argv[i], "-s") || !strcmp(argv[i], "--stdin-file")){
- 403 if(mode != MSGMODE_NONE){
- 404 fprintf(stderr, "Error: Only one type of message can be sent at once.
");
- 405 print_usage();
- 406 return 1;
- 407 }else{
- 408 if(load_stdin()) return 1;
- 409 }
- 410 }else if(!strcmp(argv[i], "-t") || !strcmp(argv[i], "--topic")){
- 411 if(i==argc-1){
- 412 fprintf(stderr, "Error: -t argument given but no topic specified.
");
- 413 print_usage();
- 414 return 1;
- 415 }else{
- 416 topic = argv[i+1];
- 417 }
- 418 i++;
- 419 }else if(!strcmp(argv[i], "-u") || !strcmp(argv[i], "--username")){
- 420 if(i==argc-1){
- 421 fprintf(stderr, "Error: -u argument given but no username specified.
");
- 422 print_usage();
- 423 return 1;
- 424 }else{
- 425 username = argv[i+1];
- 426 }
- 427 i++;
- 428 }else if(!strcmp(argv[i], "-P") || !strcmp(argv[i], "--pw")){
- 429 if(i==argc-1){
- 430 fprintf(stderr, "Error: -P argument given but no password specified.
");
- 431 print_usage();
- 432 return 1;
- 433 }else{
- 434 password = argv[i+1];
- 435 }
- 436 i++;
- 437 }else if(!strcmp(argv[i], "--will-payload")){
- 438 if(i==argc-1){
- 439 fprintf(stderr, "Error: --will-payload argument given but no will payload specified.
");
- 440 print_usage();
- 441 return 1;
- 442 }else{
- 443 will_payload = (uint8_t *)argv[i+1];
- 444 will_payloadlen = strlen((char *)will_payload);
- 445 }
- 446 i++;
- 447 }else if(!strcmp(argv[i], "--will-qos")){
- 448 if(i==argc-1){
- 449 fprintf(stderr, "Error: --will-qos argument given but no will QoS specified.
");
- 450 print_usage();
- 451 return 1;
- 452 }else{
- 453 will_qos = atoi(argv[i+1]);
- 454 if(will_qos < 0 || will_qos > 2){
- 455 fprintf(stderr, "Error: Invalid will QoS %d.
", will_qos);
- 456 return 1;
- 457 }
- 458 }
- 459 i++;
- 460 }else if(!strcmp(argv[i], "--will-retain")){
- 461 will_retain = true;
- 462 }else if(!strcmp(argv[i], "--will-topic")){
- 463 if(i==argc-1){
- 464 fprintf(stderr, "Error: --will-topic argument given but no will topic specified.
");
- 465 print_usage();
- 466 return 1;
- 467 }else{
- 468 will_topic = argv[i+1];
- 469 }
- 470 i++;
- 471 }else{
- 472 fprintf(stderr, "Error: Unknown option '%s'.
",argv[i]);
- 473 print_usage();
- 474 return 1;
- 475 }
- 476 }
- 477 if(id_prefix){
- 478 id = malloc(strlen(id_prefix)+10);
- 479 if(!id){
- 480 if(!quiet) fprintf(stderr, "Error: Out of memory.
");
- 481 return 1;
- 482 }
- 483 snprintf(id, strlen(id_prefix)+10, "%s%d", id_prefix, getpid());
- 484 }else if(!id){
- 485 id = malloc(30);
- 486 if(!id){
- 487 if(!quiet) fprintf(stderr, "Error: Out of memory.
");
- 488 return 1;
- 489 }
- 490 memset(hostname, 0, 21);
- 491 gethostname(hostname, 20);
- 492 snprintf(id, 23, "mosq_pub_%d_%s", getpid(), hostname);
- 493 }
- 494
- 495 if(!topic || mode == MSGMODE_NONE){
- 496 fprintf(stderr, "Error: Both topic and message must be supplied.
");
- 497 print_usage();
- 498 return 1;
- 499 }
- 500
- 501 if(will_payload && !will_topic){
- 502 fprintf(stderr, "Error: Will payload given, but no will topic given.
");
- 503 print_usage();
- 504 return 1;
- 505 }
- 506 if(will_retain && !will_topic){
- 507 fprintf(stderr, "Error: Will retain given, but no will topic given.
");
- 508 print_usage();
- 509 return 1;
- 510 }
- 511 if(password && !username){
- 512 if(!quiet) fprintf(stderr, "Warning: Not using password since username not set.
");
- 513 }
- 514 mosquitto_lib_init();
- 515 mosq = mosquitto_new(id, NULL);
- 516 if(!mosq){
- 517 if(!quiet) fprintf(stderr, "Error: Out of memory.
");
- 518 return 1;
- 519 }
- 520 if(debug){
- 521 mosquitto_log_init(mosq, MOSQ_LOG_DEBUG | MOSQ_LOG_ERR | MOSQ_LOG_WARNING
- 522 | MOSQ_LOG_NOTICE | MOSQ_LOG_INFO, MOSQ_LOG_STDERR);
- 523 }
- 524 if(will_topic && mosquitto_will_set(mosq, true, will_topic, will_payloadlen, will_payload, will_qos, will_retain)){
- 525 if(!quiet) fprintf(stderr, "Error: Problem setting will.
");
- 526 return 1;
- 527 }
- 528 if(username && mosquitto_username_pw_set(mosq, username, password)){
- 529 if(!quiet) fprintf(stderr, "Error: Problem setting username and password.
");
- 530 return 1;
- 531 }
- 532
- 533 mosquitto_connect_callback_set(mosq, my_connect_callback);
- 534 mosquitto_disconnect_callback_set(mosq, my_disconnect_callback);
- 535 mosquitto_publish_callback_set(mosq, my_publish_callback);
- 536
- 537 rc = mosquitto_connect(mosq, host, port, keepalive, true);
- 538 if(rc){
- 539
- 540 if(!quiet){
- 541 if(rc == MOSQ_ERR_ERRNO){
- 542 #ifndef WIN32
- 543 strerror_r(errno, err, 1024);
- 544 #else
- 545 FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM, NULL, errno, 0, (LPTSTR)&err, 1024, NULL);
- 546 #endif
- 547 fprintf(stderr, "Error: %s
", err);
- 548 }else{
- 549 fprintf(stderr, "Unable to connect (%d).
", rc);
- 550 }
- 551 }
- 552 return rc;
- 553 }
- 554
- 555 do{
- 556 if(mode == MSGMODE_STDIN_LINE && status == STATUS_CONNACK_RECVD){
- 557 if(fgets(buf, 1024, stdin)){
- 558 buf[strlen(buf)-1] = ' ';
- 559 rc2 = mosquitto_publish(mosq, &mid_sent, topic, strlen(buf), (uint8_t *)buf, qos, retain);
- 560 if(rc2){
- 561 if(!quiet) fprintf(stderr, "Error: Publish returned %d, disconnecting.
", rc2);
- 562 mosquitto_disconnect(mosq);
- 563 }
- 564 }else if(feof(stdin) && disconnect_sent == false){
- 565 mosquitto_disconnect(mosq);
- 566 disconnect_sent = true;
- 567 }
- 568 }
- 569 rc = mosquitto_loop(mosq, -1);
- 570 }while(rc == MOSQ_ERR_SUCCESS && connected);
- 571
- 572 if(message && mode == MSGMODE_FILE){
- 573 free(message);
- 574 }
- 575 mosquitto_destroy(mosq);
- 576 mosquitto_lib_cleanup();
- 577 return rc;
- 578 }
原文地址:https://www.cnblogs.com/dpf-learn/p/6963567.html