Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Merge pull request #84 from jsocol/filtered-heartbeats_84

pubsub_filtered: pass through heartbeats
  • Loading branch information...
commit 980ab98f440c136f8d0435abc9ca4b51f7195971 2 parents e677e0b + f870b71
@jehiah jehiah authored
Showing with 17 additions and 5 deletions.
  1. +2 −0  pubsub_filtered/README.md
  2. +15 −5 pubsub_filtered/pubsub_filtered.c
View
2  pubsub_filtered/README.md
@@ -4,6 +4,8 @@ pubsub_filtered
pubsub_filtered connects to a remote pubsub server and filters out or
hashes entries before re-publishing as a pubsub stream
+will not filter heartbeat messages but will pass them through to all clients
+
if you have a message like '{desc:"ip added", ip:"127.0.0.1"}' to encrypt the ip you would start pubsub_filtered with '-e ip'
OPTIONS
View
20 pubsub_filtered/pubsub_filtered.c
@@ -17,7 +17,6 @@
#include "pcre.h"
-#define DEBUG 1
#define SUCCESS 0
#define FAILURE 1
#define VERSION "1.3"
@@ -179,6 +178,7 @@ void process_message_cb(char *source, void *arg)
const char *raw_string;
char *encrypted_string;
const char *json_out;
+ int is_heartbeat = 0; // FALSE
struct cli *client;
struct filter *fltr;
char *subject;
@@ -195,13 +195,21 @@ void process_message_cb(char *source, void *arg)
return;
}
+ // some streams might have a heartbeat message, pass these through
+ if (json_object_object_get(json_in, "_heartbeat_") != NULL) {
+#ifdef DEBUG
+ fprintf(stdout, "heartbeat received\n");
+#endif
+ is_heartbeat = 1;
+ }
+
// filter
- if (expected_value && !filter_message_simple(expected_key, expected_value, json_in)) {
+ if (!is_heartbeat && expected_value && !filter_message_simple(expected_key, expected_value, json_in)) {
json_object_put(json_in);
return;
}
- if (expected_value_regex && !filter_message(expected_key, expected_value_regex, json_in)) {
+ if (!is_heartbeat && expected_value_regex && !filter_message(expected_key, expected_value_regex, json_in)) {
json_object_put(json_in);
return;
}
@@ -213,7 +221,9 @@ void process_message_cb(char *source, void *arg)
encrypt_fields(encrypted_fields, num_encrypted_fields, json_in);
json_out = json_object_to_json_string(json_in);
- //if (DEBUG)fprintf(stdout, "json_out = %d bytes\n" , strlen(json_out));
+#ifdef DEBUG
+ fprintf(stdout, "json_out = %d bytes\n" , strlen(json_out));
+#endif
// loop over the clients and send each this message
TAILQ_FOREACH(client, &clients, entries) {
@@ -229,7 +239,7 @@ void process_message_cb(char *source, void *arg)
continue;
}
// filter
- if (client->fltr.ok && !filter_message(client->fltr.subject, client->fltr.re, json_in)) {
+ if (!is_heartbeat && client->fltr.ok && !filter_message(client->fltr.subject, client->fltr.re, json_in)) {
continue;
}
if (client->websocket) {
Please sign in to comment.
Something went wrong with that request. Please try again.