my $callback = sub {
        my ($message, $guard) = @_;
        my $hashref = parse($message);
        if ($hashref && ($hashref->{schema} eq 'storage')) {
             my $key = kval_key($hashref->{tablename}, $hashref->{row_data});
             my $value = encode_json($hashref);
             my $resp = $kvstore->write($key, $value);
             # TODO:  Handle errors, see issue #77
         }
         undef $guard;
     };