MQTTPersistenceDefault.c
Go to the documentation of this file.
1 /*******************************************************************************
2  * Copyright (c) 2009, 2020 IBM Corp.
3  *
4  * All rights reserved. This program and the accompanying materials
5  * are made available under the terms of the Eclipse Public License v2.0
6  * and Eclipse Distribution License v1.0 which accompany this distribution.
7  *
8  * The Eclipse Public License is available at
9  * https://www.eclipse.org/legal/epl-2.0/
10  * and the Eclipse Distribution License is available at
11  * http://www.eclipse.org/org/documents/edl-v10.php.
12  *
13  * Contributors:
14  * Ian Craggs - initial API and implementation and/or initial documentation
15  * Ian Craggs - async client updates
16  * Ian Craggs - fix for bug 484496
17  * Ian Craggs - fix for issue 285
18  *******************************************************************************/
19 
31 #if !defined(NO_PERSISTENCE)
32 
33 #include "OsWrapper.h"
34 
35 #include <stdio.h>
36 #include <string.h>
37 #include <errno.h>
38 
39 #if defined(_WIN32) || defined(_WIN64)
40  #include <direct.h>
41  /* Windows doesn't have strtok_r, so remap it to strtok */
42  #define strtok_r( A, B, C ) strtok( A, B )
43  int keysWin32(char *, char ***, int *);
44  int clearWin32(char *);
45  int containskeyWin32(char *, char *);
46 #else
47  #include <sys/stat.h>
48  #include <dirent.h>
49  #include <unistd.h>
50  int keysUnix(char *, char ***, int *);
51  int clearUnix(char *);
52  int containskeyUnix(char *, char *);
53 #endif
54 
55 #include "MQTTClientPersistence.h"
56 #include "MQTTPersistenceDefault.h"
57 #include "StackTrace.h"
58 #include "Heap.h"
59 
64 int pstopen(void **handle, const char* clientID, const char* serverURI, void* context)
65 {
66  int rc = 0;
67  char *dataDir = context;
68  char *clientDir;
69  char *pToken = NULL;
70  char *save_ptr = NULL;
71  char *pCrtDirName = NULL;
72  char *pTokDirName = NULL;
73  char *perserverURI = NULL, *ptraux;
74 
75  FUNC_ENTRY;
76  /* Note that serverURI=address:port, but ":" not allowed in Windows directories */
77  if ((perserverURI = malloc(strlen(serverURI) + 1)) == NULL)
78  {
79  rc = PAHO_MEMORY_ERROR;
80  goto exit;
81  }
82  strcpy(perserverURI, serverURI);
83  while ((ptraux = strstr(perserverURI, ":")) != NULL)
84  *ptraux = '-' ;
85 
86  /* consider '/' + '-' + '\0' */
87  clientDir = malloc(strlen(dataDir) + strlen(clientID) + strlen(perserverURI) + 3);
88  if (!clientDir)
89  {
90  free(perserverURI);
91  rc = PAHO_MEMORY_ERROR;
92  goto exit;
93  }
94  sprintf(clientDir, "%s/%s-%s", dataDir, clientID, perserverURI);
95 
96 
97  /* create clientDir directory */
98 
99  /* pCrtDirName - holds the directory name we are currently trying to create. */
100  /* This gets built up level by level untipwdl the full path name is created.*/
101  /* pTokDirName - holds the directory name that gets used by strtok. */
102  if ((pCrtDirName = (char*)malloc(strlen(clientDir) + 1)) == NULL)
103  {
104  free(clientDir);
105  free(perserverURI);
106  rc = PAHO_MEMORY_ERROR;
107  goto exit;
108  }
109  if ((pTokDirName = (char*)malloc( strlen(clientDir) + 1 )) == NULL)
110  {
111  free(pCrtDirName);
112  free(clientDir);
113  free(perserverURI);
114  rc = PAHO_MEMORY_ERROR;
115  goto exit;
116  }
117  strcpy( pTokDirName, clientDir );
118 
119  /* If first character is directory separator, make sure it's in the created directory name #285 */
120  if (*pTokDirName == '/' || *pTokDirName == '\\')
121  {
122  *pCrtDirName = *pTokDirName;
123  pToken = strtok_r( pTokDirName + 1, "\\/", &save_ptr );
124  strcpy( pCrtDirName + 1, pToken );
125  }
126  else
127  {
128  pToken = strtok_r( pTokDirName, "\\/", &save_ptr );
129  strcpy( pCrtDirName, pToken );
130  }
131 
132  rc = pstmkdir( pCrtDirName );
133  pToken = strtok_r( NULL, "\\/", &save_ptr );
134  while ( (pToken != NULL) && (rc == 0) )
135  {
136  /* Append the next directory level and try to create it */
137  strcat( pCrtDirName, "/" );
138  strcat( pCrtDirName, pToken );
139  rc = pstmkdir( pCrtDirName );
140  pToken = strtok_r( NULL, "\\/", &save_ptr );
141  }
142 
143  *handle = clientDir;
144 
145  free(pTokDirName);
146  free(pCrtDirName);
147  free(perserverURI);
148 
149 exit:
150  FUNC_EXIT_RC(rc);
151  return rc;
152 }
153 
157 int pstmkdir( char *pPathname )
158 {
159  int rc = 0;
160 
161  FUNC_ENTRY;
162 #if defined(_WIN32) || defined(_WIN64)
163  if ( _mkdir( pPathname ) != 0 )
164  {
165 #else
166  /* Create a directory with read, write and execute access for the owner and read access for the group */
167 #if !defined(_WRS_KERNEL)
168  if ( mkdir( pPathname, S_IRWXU | S_IRGRP ) != 0 )
169 #else
170  if ( mkdir( pPathname ) != 0 )
171 #endif /* !defined(_WRS_KERNEL) */
172  {
173 #endif
174  if ( errno != EEXIST )
175  rc = MQTTCLIENT_PERSISTENCE_ERROR;
176  }
177 
178  FUNC_EXIT_RC(rc);
179  return rc;
180 }
181 
182 
183 
187 int pstput(void* handle, char* key, int bufcount, char* buffers[], int buflens[])
188 {
189  int rc = 0;
190  char *clientDir = handle;
191  char *file;
192  FILE *fp;
193  size_t bytesWritten = 0,
194  bytesTotal = 0;
195  int i;
196 
197  FUNC_ENTRY;
198  if (clientDir == NULL)
199  {
200  rc = MQTTCLIENT_PERSISTENCE_ERROR;
201  goto exit;
202  }
203 
204  /* consider '/' + '\0' */
205  file = malloc(strlen(clientDir) + strlen(key) + strlen(MESSAGE_FILENAME_EXTENSION) + 2 );
206  if (!file)
207  {
208  rc = PAHO_MEMORY_ERROR;
209  goto exit;
210  }
211  sprintf(file, "%s/%s%s", clientDir, key, MESSAGE_FILENAME_EXTENSION);
212 
213  fp = fopen(file, "wb");
214  if ( fp != NULL )
215  {
216  for(i=0; i<bufcount; i++)
217  {
218  bytesTotal += buflens[i];
219  bytesWritten += fwrite(buffers[i], sizeof(char), buflens[i], fp );
220  }
221  fclose(fp);
222  fp = NULL;
223  } else
224  rc = MQTTCLIENT_PERSISTENCE_ERROR;
225 
226  if (bytesWritten != bytesTotal)
227  {
228  pstremove(handle, key);
229  rc = MQTTCLIENT_PERSISTENCE_ERROR;
230  }
231 
232  free(file);
233 
234 exit:
235  FUNC_EXIT_RC(rc);
236  return rc;
237 };
238 
239 
243 int pstget(void* handle, char* key, char** buffer, int* buflen)
244 {
245  int rc = 0;
246  FILE *fp = NULL;
247  char *clientDir = handle;
248  char *filename = NULL;
249  char *buf = NULL;
250  unsigned long fileLen = 0;
251  unsigned long bytesRead = 0;
252 
253  FUNC_ENTRY;
254  if (clientDir == NULL)
255  {
256  rc = MQTTCLIENT_PERSISTENCE_ERROR;
257  goto exit;
258  }
259 
260  /* consider '/' + '\0' */
261  filename = malloc(strlen(clientDir) + strlen(key) + strlen(MESSAGE_FILENAME_EXTENSION) + 2);
262  if (!filename)
263  {
264  rc = PAHO_MEMORY_ERROR;
265  goto exit;
266  }
267  sprintf(filename, "%s/%s%s", clientDir, key, MESSAGE_FILENAME_EXTENSION);
268  fp = fopen(filename, "rb");
269  free(filename);
270  if (fp != NULL)
271  {
272  fseek(fp, 0, SEEK_END);
273  fileLen = ftell(fp);
274  fseek(fp, 0, SEEK_SET);
275  if ((buf = (char *)malloc(fileLen)) == NULL)
276  {
277  rc = PAHO_MEMORY_ERROR;
278  goto exit;
279  }
280  bytesRead = (int)fread(buf, sizeof(char), fileLen, fp);
281  *buffer = buf;
282  *buflen = bytesRead;
283  if ( bytesRead != fileLen )
284  rc = MQTTCLIENT_PERSISTENCE_ERROR;
285  fclose(fp);
286  } else
287  rc = MQTTCLIENT_PERSISTENCE_ERROR;
288 
289  /* the caller must free buf */
290 exit:
291  FUNC_EXIT_RC(rc);
292  return rc;
293 }
294 
295 
296 
300 int pstremove(void* handle, char* key)
301 {
302  int rc = 0;
303  char *clientDir = handle;
304  char *file;
305 
306  FUNC_ENTRY;
307  if (clientDir == NULL)
308  {
309  rc = MQTTCLIENT_PERSISTENCE_ERROR;
310  goto exit;
311  }
312 
313  /* consider '/' + '\0' */
314  file = malloc(strlen(clientDir) + strlen(key) + strlen(MESSAGE_FILENAME_EXTENSION) + 2);
315  if (!file)
316  {
317  rc = PAHO_MEMORY_ERROR;
318  goto exit;
319  }
320  sprintf(file, "%s/%s%s", clientDir, key, MESSAGE_FILENAME_EXTENSION);
321 
322 #if defined(_WIN32) || defined(_WIN64)
323  if ( _unlink(file) != 0 )
324  {
325 #else
326  if ( unlink(file) != 0 )
327  {
328 #endif
329  if ( errno != ENOENT )
330  rc = MQTTCLIENT_PERSISTENCE_ERROR;
331  }
332 
333  free(file);
334 
335 exit:
336  FUNC_EXIT_RC(rc);
337  return rc;
338 }
339 
340 
344 int pstclose(void* handle)
345 {
346  int rc = 0;
347  char *clientDir = handle;
348 
349  FUNC_ENTRY;
350  if (clientDir == NULL)
351  {
352  rc = MQTTCLIENT_PERSISTENCE_ERROR;
353  goto exit;
354  }
355 
356 #if defined(_WIN32) || defined(_WIN64)
357  if ( _rmdir(clientDir) != 0 )
358  {
359 #else
360  if ( rmdir(clientDir) != 0 )
361  {
362 #endif
363  if ( errno != ENOENT && errno != ENOTEMPTY )
364  rc = MQTTCLIENT_PERSISTENCE_ERROR;
365  }
366 
367  free(clientDir);
368 
369 exit:
370  FUNC_EXIT_RC(rc);
371  return rc;
372 }
373 
374 
378 int pstcontainskey(void *handle, char *key)
379 {
380  int rc = 0;
381  char *clientDir = handle;
382 
383  FUNC_ENTRY;
384  if (clientDir == NULL)
385  {
386  rc = MQTTCLIENT_PERSISTENCE_ERROR;
387  goto exit;
388  }
389 
390 #if defined(_WIN32) || defined(_WIN64)
391  rc = containskeyWin32(clientDir, key);
392 #else
393  rc = containskeyUnix(clientDir, key);
394 #endif
395 
396 exit:
397  FUNC_EXIT_RC(rc);
398  return rc;
399 }
400 
401 
402 #if defined(_WIN32) || defined(_WIN64)
403 int containskeyWin32(char *dirname, char *key)
404 {
405  int notFound = MQTTCLIENT_PERSISTENCE_ERROR;
406  int fFinished = 0;
407  char *filekey, *ptraux;
408  char dir[MAX_PATH+1];
409  WIN32_FIND_DATAA FileData;
410  HANDLE hDir;
411 
412  FUNC_ENTRY;
413  sprintf(dir, "%s/*", dirname);
414 
415  hDir = FindFirstFileA(dir, &FileData);
416  if (hDir != INVALID_HANDLE_VALUE)
417  {
418  while (!fFinished)
419  {
420  if (FileData.dwFileAttributes & FILE_ATTRIBUTE_ARCHIVE)
421  {
422  if ((filekey = malloc(strlen(FileData.cFileName) + 1)) == NULL)
423  {
424  notFound = PAHO_MEMORY_ERROR;
425  goto exit;
426  }
427  strcpy(filekey, FileData.cFileName);
428  ptraux = strstr(filekey, MESSAGE_FILENAME_EXTENSION);
429  if ( ptraux != NULL )
430  *ptraux = '\0' ;
431  if(strcmp(filekey, key) == 0)
432  {
433  notFound = 0;
434  fFinished = 1;
435  }
436  free(filekey);
437  }
438  if (!FindNextFileA(hDir, &FileData))
439  {
440  if (GetLastError() == ERROR_NO_MORE_FILES)
441  fFinished = 1;
442  }
443  }
444  FindClose(hDir);
445  }
446 exit:
447  FUNC_EXIT_RC(notFound);
448  return notFound;
449 }
450 #else
451 int containskeyUnix(char *dirname, char *key)
452 {
453  int notFound = MQTTCLIENT_PERSISTENCE_ERROR;
454  char *filekey, *ptraux;
455  DIR *dp = NULL;
456  struct dirent *dir_entry;
457  struct stat stat_info;
458 
459  FUNC_ENTRY;
460  if((dp = opendir(dirname)) != NULL)
461  {
462  while((dir_entry = readdir(dp)) != NULL && notFound)
463  {
464  char* filename = malloc(strlen(dirname) + strlen(dir_entry->d_name) + 2);
465 
466  if (!filename)
467  {
468  notFound = PAHO_MEMORY_ERROR;
469  goto exit;
470  }
471  sprintf(filename, "%s/%s", dirname, dir_entry->d_name);
472  lstat(filename, &stat_info);
473  free(filename);
474  if(S_ISREG(stat_info.st_mode))
475  {
476  if ((filekey = malloc(strlen(dir_entry->d_name) + 1)) == NULL)
477  {
478  notFound = PAHO_MEMORY_ERROR;
479  goto exit;
480  }
481  strcpy(filekey, dir_entry->d_name);
482  ptraux = strstr(filekey, MESSAGE_FILENAME_EXTENSION);
483  if ( ptraux != NULL )
484  *ptraux = '\0' ;
485  if(strcmp(filekey, key) == 0)
486  notFound = 0;
487  free(filekey);
488  }
489  }
490  }
491 
492 exit:
493  if (dp)
494  closedir(dp);
495  FUNC_EXIT_RC(notFound);
496  return notFound;
497 }
498 #endif
499 
500 
504 int pstclear(void *handle)
505 {
506  int rc = 0;
507  char *clientDir = handle;
508 
509  FUNC_ENTRY;
510  if (clientDir == NULL)
511  {
512  rc = MQTTCLIENT_PERSISTENCE_ERROR;
513  goto exit;
514  }
515 
516 #if defined(_WIN32) || defined(_WIN64)
517  rc = clearWin32(clientDir);
518 #else
519  rc = clearUnix(clientDir);
520 #endif
521 
522 exit:
523  FUNC_EXIT_RC(rc);
524  return rc;
525 }
526 
527 
528 #if defined(_WIN32) || defined(_WIN64)
529 int clearWin32(char *dirname)
530 {
531  int rc = 0;
532  int fFinished = 0;
533  char *file;
534  char dir[MAX_PATH+1];
535  WIN32_FIND_DATAA FileData;
536  HANDLE hDir;
537 
538  FUNC_ENTRY;
539  sprintf(dir, "%s/*", dirname);
540 
541  hDir = FindFirstFileA(dir, &FileData);
542  if (hDir != INVALID_HANDLE_VALUE)
543  {
544  while (!fFinished)
545  {
546  if (FileData.dwFileAttributes & FILE_ATTRIBUTE_ARCHIVE)
547  {
548  file = malloc(strlen(dirname) + strlen(FileData.cFileName) + 2);
549  if (!file)
550  {
551  rc = PAHO_MEMORY_ERROR;
552  goto exit;
553  }
554  sprintf(file, "%s/%s", dirname, FileData.cFileName);
555  rc = remove(file);
556  free(file);
557  if ( rc != 0 )
558  {
559  rc = MQTTCLIENT_PERSISTENCE_ERROR;
560  break;
561  }
562  }
563  if (!FindNextFileA(hDir, &FileData))
564  {
565  if (GetLastError() == ERROR_NO_MORE_FILES)
566  fFinished = 1;
567  }
568  }
569  FindClose(hDir);
570  } else
571  rc = MQTTCLIENT_PERSISTENCE_ERROR;
572 
573 exit:
574  FUNC_EXIT_RC(rc);
575  return rc;
576 }
577 #else
578 int clearUnix(char *dirname)
579 {
580  int rc = 0;
581  DIR *dp;
582  struct dirent *dir_entry;
583  struct stat stat_info;
584 
585  FUNC_ENTRY;
586  if((dp = opendir(dirname)) != NULL)
587  {
588  while((dir_entry = readdir(dp)) != NULL && rc == 0)
589  {
590  if (lstat(dir_entry->d_name, &stat_info) == 0 && S_ISREG(stat_info.st_mode))
591  {
592  if (remove(dir_entry->d_name) != 0 && errno != ENOENT)
593  rc = MQTTCLIENT_PERSISTENCE_ERROR;
594  }
595  }
596  closedir(dp);
597  } else
598  rc = MQTTCLIENT_PERSISTENCE_ERROR;
599 
600  FUNC_EXIT_RC(rc);
601  return rc;
602 }
603 #endif
604 
605 
609 int pstkeys(void *handle, char ***keys, int *nkeys)
610 {
611  int rc = 0;
612  char *clientDir = handle;
613 
614  FUNC_ENTRY;
615  if (clientDir == NULL)
616  {
617  rc = MQTTCLIENT_PERSISTENCE_ERROR;
618  goto exit;
619  }
620 
621 #if defined(_WIN32) || defined(_WIN64)
622  rc = keysWin32(clientDir, keys, nkeys);
623 #else
624  rc = keysUnix(clientDir, keys, nkeys);
625 #endif
626 
627 exit:
628  FUNC_EXIT_RC(rc);
629  return rc;
630 }
631 
632 
633 #if defined(_WIN32) || defined(_WIN64)
634 int keysWin32(char *dirname, char ***keys, int *nkeys)
635 {
636  int rc = 0;
637  char **fkeys = NULL;
638  int nfkeys = 0;
639  char dir[MAX_PATH+1];
640  WIN32_FIND_DATAA FileData;
641  HANDLE hDir;
642  int fFinished = 0;
643  char *ptraux;
644  int i;
645 
646  FUNC_ENTRY;
647  sprintf(dir, "%s/*", dirname);
648 
649  /* get number of keys */
650  hDir = FindFirstFileA(dir, &FileData);
651  if (hDir != INVALID_HANDLE_VALUE)
652  {
653  while (!fFinished)
654  {
655  if (FileData.dwFileAttributes & FILE_ATTRIBUTE_ARCHIVE)
656  nfkeys++;
657  if (!FindNextFileA(hDir, &FileData))
658  {
659  if (GetLastError() == ERROR_NO_MORE_FILES)
660  fFinished = 1;
661  }
662  }
663  FindClose(hDir);
664  } else
665  {
666  rc = MQTTCLIENT_PERSISTENCE_ERROR;
667  goto exit;
668  }
669 
670  if (nfkeys != 0)
671  {
672  if ((fkeys = (char **)malloc(nfkeys * sizeof(char *))) == NULL)
673  {
674  rc = PAHO_MEMORY_ERROR;
675  goto exit;
676  }
677  }
678 
679  /* copy the keys */
680  hDir = FindFirstFileA(dir, &FileData);
681  if (hDir != INVALID_HANDLE_VALUE)
682  {
683  fFinished = 0;
684  i = 0;
685  while (!fFinished)
686  {
687  if (FileData.dwFileAttributes & FILE_ATTRIBUTE_ARCHIVE)
688  {
689  if ((fkeys[i] = malloc(strlen(FileData.cFileName) + 1)) == NULL)
690  {
691  rc = PAHO_MEMORY_ERROR;
692  goto exit;
693  }
694  strcpy(fkeys[i], FileData.cFileName);
695  ptraux = strstr(fkeys[i], MESSAGE_FILENAME_EXTENSION);
696  if ( ptraux != NULL )
697  *ptraux = '\0' ;
698  i++;
699  }
700  if (!FindNextFileA(hDir, &FileData))
701  {
702  if (GetLastError() == ERROR_NO_MORE_FILES)
703  fFinished = 1;
704  }
705  }
706  FindClose(hDir);
707  } else
708  {
709  rc = MQTTCLIENT_PERSISTENCE_ERROR;
710  goto exit;
711  }
712 
713  *nkeys = nfkeys;
714  *keys = fkeys;
715  /* the caller must free keys */
716 
717 exit:
718  FUNC_EXIT_RC(rc);
719  return rc;
720 }
721 #else
722 int keysUnix(char *dirname, char ***keys, int *nkeys)
723 {
724  int rc = 0;
725  char **fkeys = NULL;
726  int nfkeys = 0;
727  char *ptraux;
728  int i;
729  DIR *dp = NULL;
730  struct dirent *dir_entry;
731  struct stat stat_info;
732 
733  FUNC_ENTRY;
734  /* get number of keys */
735  if((dp = opendir(dirname)) != NULL)
736  {
737  while((dir_entry = readdir(dp)) != NULL)
738  {
739  char* temp = malloc(strlen(dirname)+strlen(dir_entry->d_name)+2);
740 
741  if (!temp)
742  {
743  rc = PAHO_MEMORY_ERROR;
744  goto exit;
745  }
746  sprintf(temp, "%s/%s", dirname, dir_entry->d_name);
747  if (lstat(temp, &stat_info) == 0 && S_ISREG(stat_info.st_mode))
748  nfkeys++;
749  free(temp);
750  }
751  closedir(dp);
752  dp = NULL;
753  } else
754  {
755  rc = MQTTCLIENT_PERSISTENCE_ERROR;
756  goto exit;
757  }
758 
759  if (nfkeys != 0)
760  {
761  if ((fkeys = (char **)malloc(nfkeys * sizeof(char *))) == NULL)
762  {
763  rc = PAHO_MEMORY_ERROR;
764  goto exit;
765  }
766 
767  /* copy the keys */
768  if((dp = opendir(dirname)) != NULL)
769  {
770  i = 0;
771  while((dir_entry = readdir(dp)) != NULL)
772  {
773  char* temp = malloc(strlen(dirname)+strlen(dir_entry->d_name)+2);
774 
775  if (!temp)
776  {
777  free(fkeys);
778  rc = PAHO_MEMORY_ERROR;
779  goto exit;
780  }
781  sprintf(temp, "%s/%s", dirname, dir_entry->d_name);
782  if (lstat(temp, &stat_info) == 0 && S_ISREG(stat_info.st_mode))
783  {
784  if ((fkeys[i] = malloc(strlen(dir_entry->d_name) + 1)) == NULL)
785  {
786  free(temp);
787  free(fkeys);
788  rc = PAHO_MEMORY_ERROR;
789  goto exit;
790  }
791  strcpy(fkeys[i], dir_entry->d_name);
792  ptraux = strstr(fkeys[i], MESSAGE_FILENAME_EXTENSION);
793  if ( ptraux != NULL )
794  *ptraux = '\0' ;
795  i++;
796  }
797  free(temp);
798  }
799  } else
800  {
801  rc = MQTTCLIENT_PERSISTENCE_ERROR;
802  goto exit;
803  }
804  }
805 
806  *nkeys = nfkeys;
807  *keys = fkeys;
808  /* the caller must free keys */
809 
810 exit:
811  if (dp)
812  closedir(dp);
813  FUNC_EXIT_RC(rc);
814  return rc;
815 }
816 #endif
817 
818 
819 
820 #if defined(UNIT_TESTS)
821 int main (int argc, char *argv[])
822 {
823 #define MSTEM "m-"
824 #define NMSGS 10
825 #define NBUFS 4
826 #define NDEL 2
827 #define RC !rc ? "(Success)" : "(Failed) "
828 
829  int rc;
830  char *handle;
831  char *perdir = ".";
832  const char *clientID = "TheUTClient";
833  const char *serverURI = "127.0.0.1:1883";
834 
835  char *stem = MSTEM;
836  int msgId, i;
837  int nm[NDEL] = {5 , 8}; /* msgIds to get and remove */
838 
839  char *key;
840  char **keys;
841  int nkeys;
842  char *buffer, *buff;
843  int buflen;
844 
845  int nbufs = NBUFS;
846  char *bufs[NBUFS] = {"m0", "mm1", "mmm2" , "mmmm3"}; /* message content */
847  int buflens[NBUFS];
848  for(i=0;i<nbufs;i++)
849  buflens[i]=strlen(bufs[i]);
850 
851  /* open */
852  /* printf("Persistence directory : %s\n", perdir); */
853  rc = pstopen((void**)&handle, clientID, serverURI, perdir);
854  printf("%s Persistence directory for client %s : %s\n", RC, clientID, handle);
855 
856  /* put */
857  for(msgId=0;msgId<NMSGS;msgId++)
858  {
859  key = malloc(MESSAGE_FILENAME_LENGTH + 1);
860  sprintf(key, "%s%d", stem, msgId);
861  rc = pstput(handle, key, nbufs, bufs, buflens);
862  printf("%s Adding message %s\n", RC, key);
863  free(key);
864  }
865 
866  /* keys ,ie, list keys added */
867  rc = pstkeys(handle, &keys, &nkeys);
868  printf("%s Found %d messages persisted in %s\n", RC, nkeys, handle);
869  for(i=0;i<nkeys;i++)
870  printf("%13s\n", keys[i]);
871 
872  if (keys !=NULL)
873  free(keys);
874 
875  /* containskey */
876  for(i=0;i<NDEL;i++)
877  {
878  key = malloc(MESSAGE_FILENAME_LENGTH + 1);
879  sprintf(key, "%s%d", stem, nm[i]);
880  rc = pstcontainskey(handle, key);
881  printf("%s Message %s is persisted ?\n", RC, key);
882  free(key);
883  }
884 
885  /* get && remove*/
886  for(i=0;i<NDEL;i++)
887  {
888  key = malloc(MESSAGE_FILENAME_LENGTH + 1);
889  sprintf(key, "%s%d", stem, nm[i]);
890  rc = pstget(handle, key, &buffer, &buflen);
891  buff = malloc(buflen+1);
892  memcpy(buff, buffer, buflen);
893  buff[buflen] = '\0';
894  printf("%s Retrieving message %s : %s\n", RC, key, buff);
895  rc = pstremove(handle, key);
896  printf("%s Removing message %s\n", RC, key);
897  free(key);
898  free(buff);
899  free(buffer);
900  }
901 
902  /* containskey */
903  for(i=0;i<NDEL;i++)
904  {
905  key = malloc(MESSAGE_FILENAME_LENGTH + 1);
906  sprintf(key, "%s%d", stem, nm[i]);
907  rc = pstcontainskey(handle, key);
908  printf("%s Message %s is persisted ?\n", RC, key);
909  free(key);
910  }
911 
912  /* keys ,ie, list keys added */
913  rc = pstkeys(handle, &keys, &nkeys);
914  printf("%s Found %d messages persisted in %s\n", RC, nkeys, handle);
915  for(i=0;i<nkeys;i++)
916  printf("%13s\n", keys[i]);
917 
918  if (keys != NULL)
919  free(keys);
920 
921 
922  /* close -> it will fail, since client persistence directory is not empty */
923  rc = pstclose(&handle);
924  printf("%s Closing client persistence directory for client %s\n", RC, clientID);
925 
926  /* clear */
927  rc = pstclear(handle);
928  printf("%s Deleting all persisted messages in %s\n", RC, handle);
929 
930  /* keys ,ie, list keys added */
931  rc = pstkeys(handle, &keys, &nkeys);
932  printf("%s Found %d messages persisted in %s\n", RC, nkeys, handle);
933  for(i=0;i<nkeys;i++)
934  printf("%13s\n", keys[i]);
935 
936  if ( keys != NULL )
937  free(keys);
938 
939  /* close */
940  rc = pstclose(&handle);
941  printf("%s Closing client persistence directory for client %s\n", RC, clientID);
942 }
943 #endif
944 
945 
946 #endif /* NO_PERSISTENCE */
pstopen
int pstopen(void **handle, const char *clientID, const char *serverURI, void *context)
Definition: MQTTPersistenceDefault.c:64
MQTTCLIENT_PERSISTENCE_ERROR
#define MQTTCLIENT_PERSISTENCE_ERROR
Definition: MQTTClientPersistence.h:85
RC
#define RC(i)
Definition: lvm.c:1039
pstmkdir
int pstmkdir(char *pPathname)
Definition: MQTTPersistenceDefault.c:157
PAHO_MEMORY_ERROR
#define PAHO_MEMORY_ERROR
Definition: Heap.h:26
pstclear
int pstclear(void *handle)
Definition: MQTTPersistenceDefault.c:504
malloc
#define malloc(x)
Definition: Heap.h:41
free
#define free(x)
Definition: Heap.h:55
_unlink
#define _unlink
Definition: Log.c:59
FUNC_EXIT_RC
#define FUNC_EXIT_RC(x)
Definition: StackTrace.h:63
keysUnix
int keysUnix(char *, char ***, int *)
Definition: MQTTPersistenceDefault.c:722
MQTTClientPersistence.h
This structure represents a persistent data store, used to store outbound and inbound messages...
pstcontainskey
int pstcontainskey(void *handle, char *key)
Definition: MQTTPersistenceDefault.c:378
MESSAGE_FILENAME_EXTENSION
#define MESSAGE_FILENAME_EXTENSION
Definition: MQTTPersistenceDefault.h:23
pstget
int pstget(void *handle, char *key, char **buffer, int *buflen)
Definition: MQTTPersistenceDefault.c:243
FUNC_ENTRY
#define FUNC_ENTRY
Definition: StackTrace.h:55
pstclose
int pstclose(void *handle)
Definition: MQTTPersistenceDefault.c:344
pstput
int pstput(void *handle, char *key, int bufcount, char *buffers[], int buflens[])
Definition: MQTTPersistenceDefault.c:187
test2.context
dictionary context
Definition: test2.py:57
MESSAGE_FILENAME_LENGTH
#define MESSAGE_FILENAME_LENGTH
Definition: MQTTPersistenceDefault.h:21
pstremove
int pstremove(void *handle, char *key)
Definition: MQTTPersistenceDefault.c:300
containskeyUnix
int containskeyUnix(char *, char *)
Definition: MQTTPersistenceDefault.c:451
rc
enum MQTTReasonCodes rc
Definition: test10.c:1112
main
int main(int argc, char **argv)
Definition: lua.c:619
pstkeys
int pstkeys(void *handle, char ***keys, int *nkeys)
Definition: MQTTPersistenceDefault.c:609
clearUnix
int clearUnix(char *)
Definition: MQTTPersistenceDefault.c:578

plotjuggler
Author(s): Davide Faconti
autogenerated on Sun Dec 6 2020 03:48:09