FileSpewer somewhat works now. Improvements should be made regarding the

memory allocation for the send buffer (currently it is malloc'ed every
iteration) and the message format (client Id, message topic, payload
format) should be discussed for our future work.

FileSpewer/FileSpewer.cpp

@@ -8,25 +8,137 @@
 
 #include <cstdio>
 #include <cstdlib>
+#include <csignal>
 
 #include <unistd.h>
+#include <sys/stat.h>
 
 #include "Settings.h"
 #include "mosquitto.h"
 
 Settings* mySettings;
+int keepRunning;
+
+void sigHandler(int sig) {
+  keepRunning = 0;
+}
 
 void spew() {
   int mosqMajor, mosqMinor, mosqRevision;
+  FILE *fp;
 
   /* Print mosquitto version and initialize library */
   mosquitto_lib_version(&mosqMajor, &mosqMinor, &mosqRevision);
   printf("Initializing Mosquitto Library Version %d.%d.%d\n", mosqMajor, mosqMinor, mosqRevision);
   mosquitto_lib_init();
 
+  /* Build filename as concatenation of hostname and absolute path and ensure that the file exists and is not a directory */
+  char tmpHostname[256];
+  if (gethostname(tmpHostname, 255) != 0) {
+      fprintf(stderr, "Cannot get hostname.\n");
+      exit(EXIT_FAILURE);
+  }
+  tmpHostname[255] = '\0';
+
+  char tmpFilePath[PATH_MAX];
+  if (realpath(mySettings->getFilename().c_str(), tmpFilePath) == NULL) {
+      perror(mySettings->getFilename().c_str());
+      exit(EXIT_FAILURE);
+  }
+  tmpFilePath[PATH_MAX-1] = '\0';
+
+  struct stat sbuf;
+  stat(tmpFilePath, &sbuf);
+  if (!(sbuf.st_mode & (S_IFREG|S_IFCHR))) {
+      fprintf(stderr, "%s is not a regular file!\n", tmpFilePath);
+      exit(EXIT_FAILURE);
+  }
+
+  fp = fopen(tmpFilePath, "r");
+  if(!fp) {
+      perror(tmpFilePath);
+      exit(EXIT_FAILURE);
+  }
+
+  std::string clientId;
+  clientId = tmpHostname;
+  clientId.append("/");
+  clientId.append(tmpFilePath);
+
+  printf("Client ID: %s\n", clientId.c_str());
+
+  /* Init mosquitto struct */
+  struct mosquitto* mosq;
+  mosq = mosquitto_new(clientId.c_str(), false, NULL);
+  if (!mosq) {
+      perror(NULL);
+      exit(EXIT_FAILURE);
+  }
+
+  /* Connect to the broker */
+  printf("Connecting to broker...");
+  fflush(stdout);
+  if (mosquitto_connect(mosq, mySettings->getTargetHost().c_str(), 1883, 1000) != MOSQ_ERR_SUCCESS) {
+      perror("\nCould not connect to host");
+      exit(EXIT_FAILURE);
+  }
+  printf(" Done.\n");
+
+  /* Catch SIGINT signals */
+  signal(SIGINT, sigHandler);
+
+
+  /* Here comes the main loop */
+  keepRunning = 1;
+  while(keepRunning) {
+
+      /* Determine file size */
+      off_t len;
+      if(fseek(fp, 0, SEEK_END)!=0) {
+          fprintf(stderr, "Warning: cannot determine file size.\n");
+      }
+      len = ftello(fp);
+      fseek(fp, 0, SEEK_SET);
+
+      /* Alloc send buffer (FIXME: shouldn't be doing this on every iteration) */
+      void *data = malloc(len);
+      if(!data) {
+          fprintf(stderr, "Ouch, malloc failed.\n");
+      }
+
+      /* Read data into buffer */
+      fread(data, len, 1, fp);
+
+      /* Send the message */
+      if(mosquitto_publish(mosq, NULL, tmpFilePath, len, data, 0, false) != MOSQ_ERR_SUCCESS) {
+          fprintf(stderr, "Warning: cannot send message.\n");
+      }
+      else {
+          printf("Sent %lld bytes...\n", len);
+      }
+
+      /* Free buffer */
+      free(data);
+
+      /* Wait for next iteration */
+      usleep(mySettings->getInterval() * 1000);
+  }
+
+
+  /* Disconnect from broker */
+  printf("\nDisconnecting from broker...");
+  if (mosquitto_disconnect(mosq) != MOSQ_ERR_SUCCESS) {
+      fprintf(stderr, "\nError while disconnecting!\n");
+      exit(EXIT_FAILURE);
+  }
+
+  printf(" Done.\n");
 
+  /* Close the file */
+  fclose(fp);
 
   /* Cleanup mosquitto library */
+  mosquitto_destroy(mosq);
   mosquitto_lib_cleanup();
 }
 

FileSpewer/Makefile

@@ -7,7 +7,7 @@ LIBS =		-L../../mosquitto-1.1/lib/ -lmosquitto
 TARGET =	FileSpewer
 
 $(TARGET):	$(OBJS)
-	$(CXX) -o $(TARGET) $(OBJS) $(LIBS) $(INCLUDES)
+	$(CXX) -o $(TARGET) $(OBJS) $(LIBS)
 	
 all:	$(TARGET)
 

FileSpewer/Settings.cpp

@@ -50,7 +50,8 @@ Settings::Settings(int argc, char* const argv[]) : argc_(argc), argv_(argv)
           return;
       }
 
-      clientId_ = argv[optind];
+      /* Store filename in our class */
+      filename_ = argv[optind];
 
       /* No failure, so we're good to go */
       mode_ = SPEW;
@@ -61,3 +62,18 @@ UsageMode Settings::getMode()
 {
   return mode_;
 }
+
+std::string Settings::getFilename()
+{
+  return filename_;
+}
+
+std::string Settings::getTargetHost()
+{
+  return targetHost_;
+}
+
+int Settings::getInterval()
+{
+  return interval_;
+}

FileSpewer/Settings.h

@@ -24,11 +24,14 @@ protected:
   UsageMode mode_;
   unsigned int interval_;
   std::string targetHost_;
-  std::string clientId_;
+  std::string filename_;
 
 public:
   Settings(int argc, char * const argv[]);
   UsageMode getMode();
+  std::string getFilename();
+  std::string getTargetHost();
+  int getInterval();
 
 };