d0kuhn/mosquitto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

msgps clients now report continuously.

Browse Source
This commit is contained in:
Roger A. Light 2020-10-07 17:20:50 +01:00
parent 988b5cf0b2
commit e104645279
3 changed files with 27 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
test/old/msgsps_common.h
View File

@ -1,9 +1,7 @@
#define HOST "127.0.0.1" #define HOST "127.0.0.1"
#define PORT 1888 #define PORT 1883
#define PUB_QOS 1 #define PUB_QOS 1
#define SUB_QOS 1 #define SUB_QOS 1
#define MESSAGE_COUNT 100000L
#define MESSAGE_SIZE 1024L #define MESSAGE_SIZE 1024L

61
test/old/msgsps_pub.c
View File

@ -4,76 +4,45 @@
#include <stdint.h> #include <stdint.h>
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
#include <sys/time.h> #include <unistd.h>
#include <mosquitto.h> #include <mosquitto.h>
#include <msgsps_common.h> #include <msgsps_common.h>
static bool run = true; static volatile int message_count = 0;
static int message_count = 0;
static struct timeval start, stop;
void my_connect_callback(struct mosquitto *mosq, void *obj, int rc)
{
printf("rc: %d\n", rc);
gettimeofday(&start, NULL);
}
void my_disconnect_callback(struct mosquitto *mosq, void *obj, int result)
{
run = false;
}
void my_publish_callback(struct mosquitto *mosq, void *obj, int mid) void my_publish_callback(struct mosquitto *mosq, void *obj, int mid)
{ {
message_count++; message_count++;
if(message_count == MESSAGE_COUNT){
gettimeofday(&stop, NULL);
mosquitto_disconnect((struct mosquitto *)obj);
}
} }
int main(int argc, char *argv[]) int main(int argc, char *argv[])
{ {
struct mosquitto *mosq; struct mosquitto *mosq;
int i; int i;
double dstart, dstop, diff; uint8_t buf[MESSAGE_SIZE];
uint8_t *buf;
buf = malloc(MESSAGE_SIZE*MESSAGE_COUNT);
if(!buf){
printf("Error: Out of memory.\n");
return 1;
}
start.tv_sec = 0;
start.tv_usec = 0;
stop.tv_sec = 0;
stop.tv_usec = 0;
mosquitto_lib_init(); mosquitto_lib_init();
mosq = mosquitto_new("perftest", true, NULL); mosq = mosquitto_new(NULL, true, NULL);
mosquitto_connect_callback_set(mosq, my_connect_callback);
mosquitto_disconnect_callback_set(mosq, my_disconnect_callback);
mosquitto_publish_callback_set(mosq, my_publish_callback); mosquitto_publish_callback_set(mosq, my_publish_callback);
mosquitto_connect(mosq, HOST, PORT, 600); mosquitto_connect(mosq, HOST, PORT, 600);
mosquitto_loop_start(mosq); mosquitto_loop_start(mosq);
i=0; i=0;
for(i=0; i<MESSAGE_COUNT; i++){ while(1){
mosquitto_publish(mosq, NULL, "perf/test", MESSAGE_SIZE, &buf[i*MESSAGE_SIZE], PUB_QOS, false); mosquitto_publish(mosq, NULL, "perf/test", sizeof(buf), buf, PUB_QOS, false);
usleep(100);
i++;
if(i == 10000){
/* Crude "messages per second" count */
i = message_count;
message_count = 0;
printf("%d\n", i);
i = 0;
}
} }
mosquitto_loop_stop(mosq, false); mosquitto_loop_stop(mosq, false);
dstart = (double)start.tv_sec*1.0e6 + (double)start.tv_usec;
dstop = (double)stop.tv_sec*1.0e6 + (double)stop.tv_usec;
diff = (dstop-dstart)/1.0e6;
printf("Start: %g\nStop: %g\nDiff: %g\nMessages/s: %g\n", dstart, dstop, diff, (double)MESSAGE_COUNT/diff);
mosquitto_destroy(mosq); mosquitto_destroy(mosq);
mosquitto_lib_cleanup(); mosquitto_lib_cleanup();

59
test/old/msgsps_sub.c
View File

@ -9,74 +9,37 @@
#include <msgsps_common.h> #include <msgsps_common.h>
static bool run = true; static volatile int message_count = 0;
static int message_count = 0;
static struct timeval start, stop;
FILE *fptr = NULL;
void my_connect_callback(struct mosquitto *mosq, void *obj, int rc)
{
printf("rc: %d\n", rc);
}
void my_disconnect_callback(struct mosquitto *mosq, void *obj, int result)
{
run = false;
}
void my_message_callback(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg) void my_message_callback(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg)
{ {
if(message_count == 0){
gettimeofday(&start, NULL);
}
//fwrite(msg->payload, sizeof(uint8_t), msg->payloadlen, fptr);
message_count++; message_count++;
if(message_count == MESSAGE_COUNT){
gettimeofday(&stop, NULL);
mosquitto_disconnect((struct mosquitto *)obj);
}
} }
int main(int argc, char *argv[]) int main(int argc, char *argv[])
{ {
struct mosquitto *mosq; struct mosquitto *mosq;
double dstart, dstop, diff; int c;
int mid = 0;
char id[50];
start.tv_sec = 0;
start.tv_usec = 0;
stop.tv_sec = 0;
stop.tv_usec = 0;
fptr = fopen("msgsps_sub.dat", "wb");
if(!fptr){
printf("Error: Unable to write to msgsps_sub.dat.\n");
return 1;
}
mosquitto_lib_init(); mosquitto_lib_init();
snprintf(id, 50, "msgps_sub_%d", getpid()); mosq = mosquitto_new(NULL, true, NULL);
mosq = mosquitto_new(id, true, NULL);
mosquitto_connect_callback_set(mosq, my_connect_callback);
mosquitto_disconnect_callback_set(mosq, my_disconnect_callback);
mosquitto_message_callback_set(mosq, my_message_callback); mosquitto_message_callback_set(mosq, my_message_callback);
mosquitto_connect(mosq, HOST, PORT, 600); mosquitto_connect(mosq, HOST, PORT, 600);
mosquitto_subscribe(mosq, &mid, "perf/test", SUB_QOS); mosquitto_subscribe(mosq, NULL, "perf/test", SUB_QOS);
mosquitto_loop_forever(mosq, 10, 1); mosquitto_loop_start(mosq);
while(1){
sleep(1);
c = message_count;
message_count = 0;
printf("%d\n", c);
dstart = (double)start.tv_sec*1.0e6 + (double)start.tv_usec; }
dstop = (double)stop.tv_sec*1.0e6 + (double)stop.tv_usec;
diff = (dstop-dstart)/1.0e6;
printf("Start: %g\nStop: %g\nDiff: %g\nMessages/s: %g\n", dstart, dstop, diff, (double)MESSAGE_COUNT/diff);
mosquitto_destroy(mosq); mosquitto_destroy(mosq);
mosquitto_lib_cleanup(); mosquitto_lib_cleanup();
fclose(fptr);
return 0; return 0;
} }