Use monads capabilities and more explicit variable name (#7195)

Technical refacto in order to use monads capability instead of doing a manual filter.

Rename some variables too.

Author: benmoriceau

airbyte-workers/src/main/java/io/airbyte/workers/protocols/airbyte/DefaultAirbyteStreamFactory.java

@@ -44,40 +44,36 @@ public DefaultAirbyteStreamFactory() {
   public Stream<AirbyteMessage> create(final BufferedReader bufferedReader) {
     return bufferedReader
         .lines()
-        .map(s -> {
-          final Optional<JsonNode> j = Jsons.tryDeserialize(s);
-          if (j.isEmpty()) {
+        .flatMap(line -> {
+          final Optional<JsonNode> jsonLine = Jsons.tryDeserialize(line);
+          if (jsonLine.isEmpty()) {
             // we log as info all the lines that are not valid json
             // some sources actually log their process on stdout, we
             // want to make sure this info is available in the logs.
-            logger.info(s);
+            logger.info(line);
           }
-          return j;
+          return jsonLine.stream();
         })
-        .filter(Optional::isPresent)
-        .map(Optional::get)
         // filter invalid messages
-        .filter(j -> {
-          final boolean res = protocolValidator.test(j);
+        .filter(jsonLine -> {
+          final boolean res = protocolValidator.test(jsonLine);
           if (!res) {
-            logger.error("Validation failed: {}", Jsons.serialize(j));
+            logger.error("Validation failed: {}", Jsons.serialize(jsonLine));
           }
           return res;
         })
-        .map(j -> {
-          final Optional<AirbyteMessage> m = Jsons.tryObject(j, AirbyteMessage.class);
+        .flatMap(jsonLine -> {
+          final Optional<AirbyteMessage> m = Jsons.tryObject(jsonLine, AirbyteMessage.class);
           if (m.isEmpty()) {
-            logger.error("Deserialization failed: {}", Jsons.serialize(j));
+            logger.error("Deserialization failed: {}", Jsons.serialize(jsonLine));
           }
-          return m;
+          return m.stream();
         })
-        .filter(Optional::isPresent)
-        .map(Optional::get)
         // filter logs
-        .filter(m -> {
-          final boolean isLog = m.getType() == AirbyteMessage.Type.LOG;
+        .filter(airbyteMessage -> {
+          final boolean isLog = airbyteMessage.getType() == AirbyteMessage.Type.LOG;
           if (isLog) {
-            internalLog(m.getLog());
+            internalLog(airbyteMessage.getLog());
           }
           return !isLog;
         });