yaml-dsl: enable flow mode deserialization by default

docs/modules/ROOT/pages/reference/extensions/yaml-dsl.adoc

@@ -31,3 +31,30 @@ Or add the coordinates to your existing project:
 ----
 
 Check the xref:user-guide/index.adoc[User guide] for more information about writing Camel Quarkus applications.
+
+== Additional Camel Quarkus configuration
+
+[width="100%",cols="80,5,15",options="header"]
+|===
+| Configuration property | Type | Default
+
+
+|icon:lock[title=Fixed at build time] [[quarkus.camel.yaml.flow-mode]]`link:#quarkus.camel.yaml.flow-mode[quarkus.camel.yaml.flow-mode]`
+
+If `true` the YAML DSL support flow-mode which allow to write more concise routes as for EIPs that have their own output like filter, aggregate, split, etc. the `steps` element can be omitted an in that case, the next processing step is automatically wired to the EIP's outputs. 
+ As example, a YAML DSL to process only the timer events from 5 to 10 would look like: `- from:
+    uri: "timer:tick"
+    steps:
+      - filter:
+          simple: "$++{++exchangeProperty.CamelTimerCounter` range '5..10'" steps: - to: "direct:filtered" ++}++  With the flow mode enabled the same logic can be expressed in a more concise way: `- from:
+    uri: "kamelet:source"
+    steps:
+      - filter:
+          simple: "$++{++exchangeProperty.CamelTimerCounter` range '5..10'" - to: "kamelet:sink" ++}++
+| `boolean`
+| `true`
+|===
+
+[.configuration-legend]
+icon:lock[title=Fixed at build time] Configuration property fixed at build time. All other configuration properties are overridable at runtime.
+

extensions-core/yaml-dsl/deployment/src/main/java/org/apache/camel/quarkus/dsl/yaml/deployment/YamlDslProcessor.java

@@ -18,7 +18,13 @@
 package org.apache.camel.quarkus.dsl.yaml.deployment;
 
 import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.annotations.ExecutionTime;
+import io.quarkus.deployment.annotations.Record;
 import io.quarkus.deployment.builditem.FeatureBuildItem;
+import org.apache.camel.dsl.yaml.common.YamlDeserializationMode;
+import org.apache.camel.quarkus.core.deployment.spi.CamelContextCustomizerBuildItem;
+import org.apache.camel.quarkus.dsl.yaml.YamlDslConfiguration;
+import org.apache.camel.quarkus.dsl.yaml.YamlDslRecorder;
 
 public class YamlDslProcessor {
     private static final String FEATURE = "camel-yaml-dsl";
@@ -27,4 +33,16 @@ public class YamlDslProcessor {
     FeatureBuildItem feature() {
         return new FeatureBuildItem(FEATURE);
     }
+
+    @BuildStep
+    @Record(value = ExecutionTime.STATIC_INIT)
+    CamelContextCustomizerBuildItem enableFlowMode(
+            YamlDslRecorder recorder,
+            YamlDslConfiguration configuration) {
+        return new CamelContextCustomizerBuildItem(
+                recorder.setYamlDeserializationMode(
+                        configuration.flowMode
+                                ? YamlDeserializationMode.FLOW.name()
+                                : YamlDeserializationMode.CLASSIC.name()));
+    }
 }

extensions-core/yaml-dsl/runtime/src/main/java/org/apache/camel/quarkus/dsl/yaml/YamlDslConfiguration.java

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.quarkus.dsl.yaml;
+
+import io.quarkus.runtime.annotations.ConfigItem;
+import io.quarkus.runtime.annotations.ConfigPhase;
+import io.quarkus.runtime.annotations.ConfigRoot;
+
+@ConfigRoot(name = "camel.yaml", phase = ConfigPhase.BUILD_TIME)
+public class YamlDslConfiguration {
+    /**
+     * If {@code true} the YAML DSL support flow-mode which allow to write more concise routes as for EIPs that have
+     * their own output like filter, aggregate, split, etc. the {@code steps} element can be omitted an in that case,
+     * the next processing step is automatically wired to the EIP's outputs.
+     * <p/>
+     * As example, a YAML DSL to process only the timer events from 5 to 10 would look like:
+     *
+     * <pre>
+     * {@code
+     * - from:
+     *     uri: "timer:tick"
+     *     steps:
+     *       - filter:
+     *           simple: "${exchangeProperty.CamelTimerCounter} range '5..10'"
+     *           steps:
+     *             - to: "direct:filtered"
+     * }
+     * </pre>
+     *
+     * With the flow mode enabled the same logic can be expressed in a more concise way:
+     *
+     * <pre>
+     * {@code
+     * - from:
+     *     uri: "kamelet:source"
+     *     steps:
+     *       - filter:
+     *           simple: "${exchangeProperty.CamelTimerCounter} range '5..10'"
+     *       - to: "kamelet:sink"
+     * }
+     * </pre>
+     *
+     *
+     */
+    @ConfigItem(defaultValue = "true")
+    public boolean flowMode;
+}

extensions-core/yaml-dsl/runtime/src/main/java/org/apache/camel/quarkus/dsl/yaml/YamlDslRecorder.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.camel.quarkus.dsl.yaml;
+
+import io.quarkus.runtime.RuntimeValue;
+import io.quarkus.runtime.annotations.Recorder;
+import org.apache.camel.CamelContext;
+import org.apache.camel.dsl.yaml.YamlRoutesBuilderLoader;
+import org.apache.camel.spi.CamelContextCustomizer;
+
+@Recorder
+public class YamlDslRecorder {
+    public RuntimeValue<CamelContextCustomizer> setYamlDeserializationMode(String mode) {
+        return new RuntimeValue<>(new CamelContextCustomizer() {
+            @Override
+            public void configure(CamelContext camelContext) {
+                camelContext.getGlobalOptions().put(
+                        YamlRoutesBuilderLoader.DESERIALIZATION_MODE,
+                        mode);
+            }
+        });
+    }
+}

integration-tests/main-yaml/src/main/java/org/apache/camel/quarkus/main/CoreMainYamlResource.java

@@ -16,6 +16,8 @@
  */
 package org.apache.camel.quarkus.main;
 
+import java.util.Map;
+
 import javax.enterprise.context.ApplicationScoped;
 import javax.inject.Inject;
 import javax.json.Json;
@@ -39,6 +41,7 @@ public class CoreMainYamlResource {
     @Path("/main/describe")
     @GET
     @Produces(MediaType.APPLICATION_JSON)
+    @SuppressWarnings("unchecked")
     public JsonObject describeMain() {
         final ExtendedCamelContext camelContext = main.getCamelContext().adapt(ExtendedCamelContext.class);
 
@@ -57,6 +60,7 @@ public JsonObject describeMain() {
                                 .findClass(YamlRoutesBuilderLoader.EXTENSION).get().getName())
                 .add("routeBuilders", routeBuilders)
                 .add("routes", routes)
+                .add("global-options", Json.createObjectBuilder((Map) main.getCamelContext().getGlobalOptions()).build())
                 .build();
     }
 }

integration-tests/main-yaml/src/test/java/org/apache/camel/quarkus/main/CoreMainYamlTest.java

@@ -22,6 +22,7 @@
 import io.restassured.RestAssured;
 import io.restassured.path.json.JsonPath;
 import org.apache.camel.dsl.yaml.YamlRoutesBuilderLoader;
+import org.apache.camel.dsl.yaml.common.YamlDeserializationMode;
 import org.junit.jupiter.api.Test;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -45,5 +46,7 @@ public void testMainInstanceWithYamlRoutes() {
                 .isEmpty();
         assertThat(p.getList("routes", String.class))
                 .contains("my-yaml-route", "rest-route");
+        assertThat(p.getMap("global-options", String.class, String.class))
+                .containsEntry(YamlRoutesBuilderLoader.DESERIALIZATION_MODE, YamlDeserializationMode.FLOW.name());
     }
 }