+ my $dir = $topic;
+ $dir =~ s{\w+/\w+/(\w+)/\w+}{$1};
+
+ mkdir "$queue" if ( ! -e "$queue" );
+ mkdir "$queue/$dir" if ( ! -e "$queue/$dir" );
+ mkdir "$queue/$dir/up" if ( ! -e "$queue/$dir/up" );
+ write_file "$queue/$dir/up/$t", $msg->string;
+
+ mkdir "$queue/$dir/down" if ( ! -e "$queue/$dir/down" );
+ mkdir "$queue/$dir/down/.done" if ( ! -e "$queue/$dir/down/.done" );
+
+# if ( $msg->topic =~ m{Inclinometer/ZCT330Ex_SWP_N_YK/869858031634109/up} ) {
+ if ( substr($msg->message,2,1) eq "\x07" ) { # heartbeat
+ $topic =~ s/up$/down/;
+
+ my @all_pending = sort glob "$queue/$dir/down/*";
+ if ( my $pending = shift @all_pending ) {
+ my $raw = read_file $pending;
+
+ send_message($socket,
+ message_type => MQTT_PUBLISH,
+ retain => 0, #$retain,
+ topic => $topic,
+ message => $raw);
+ rename "$queue/$dir/down/$pending", "$queue/$dir/down/.done/$pending";
+ }