journal-remote: when an entry is rejected with -EBADMSG, do not rotate the journal file

Something is wrong with the entry (probably a missing timestamp), so no point
in rotating. But suppress the error in process_source(), so that the processing
of the data stream continues.

Also, just return 0 from writer_write() on success, the only caller doesn't
care.

src/journal-remote/journal-remote-parse.c

@@ -32,7 +32,6 @@ void source_free(RemoteSource *source) {
  * ownership of fd, name, and writer, otherwise does not touch them.
  */
 RemoteSource* source_new(int fd, bool passive_fd, char *name, Writer *writer) {
-
         RemoteSource *source;
 
         log_debug("Creating source for %sfd:%d (%s)",
@@ -75,7 +74,10 @@ int process_source(RemoteSource *source, bool compress, bool seal) {
         assert(source->importer.iovw.iovec);
 
         r = writer_write(source->writer, &source->importer.iovw, &source->importer.ts, compress, seal);
-        if (r < 0)
+        if (r == -EBADMSG) {
+                log_error_errno(r, "Entry is invalid, ignoring.");
+                r = 0;
+        } else if (r < 0)
                 log_error_errno(r, "Failed to write entry of %zu bytes: %m",
                                 iovw_size(&source->importer.iovw));
         else

src/journal-remote/journal-remote-write.c

@@ -97,8 +97,9 @@ int writer_write(Writer *w,
         if (r >= 0) {
                 if (w->server)
                         w->server->event_count += 1;
-                return 1;
-        }
+                return 0;
+        } else if (r == -EBADMSG)
+                return r;
 
         log_debug_errno(r, "%s: Write failed, rotating: %m", w->journal->path);
         r = do_rotate(&w->journal, compress, seal);
@@ -115,5 +116,5 @@ int writer_write(Writer *w,
 
         if (w->server)
                 w->server->event_count += 1;
-        return 1;
+        return 0;
 }