Enhance mongodb sink kamelet

Adding more parameters and expected headers to control document
filtering, sorting, limits, skip, etc

mongodb-sink.kamelet.yaml

@@ -17,12 +17,25 @@ spec:
       Send documents to MongoDB.
 
       This Kamelet expects a JSON as body.
+      If the `operation` parameter is not set, use `insert` as default.
+
+      There are several properties you can set in the headers, such as:
+
+      `criteria` / `ce-criteria` Query in JSON format to use in some mongodb operations to find documents in MongoDB. JSON String value. Example: "{$and : [{\"storeLocation\": \"Denver\"}, {\"customer.age\": {\"$gte\": 60}}]}";
+      `num-skip` / `ce-num-skip` Discards a given number of elements at the beginning of the cursor. Integer value.
+      `limit` / `ce-limit` Limits the number of elements returned. Integer value.
+      `batch-size` / `ce-batch-size` Limits the number of elements returned in one batch. Integer value.
+      `sort-by` / `ce-sort-by` Sorts the result by field and 1 for ascending or -1 for descending order. String value in json syntax. Example "{storeLocation: 1}"
+      `multi-update` / `ce-multi-update` In an update operation, MongoDB will update only one object, you can set this parameter to true to update all matching objects. Boolean value.
+      `db-upsert` / `ce-db-upsert` If the database should create the element if it does not exist. Boolean value.
+      `collection-index` / `ce-collection-index` Sets the collection index in JSON format. String value.
     required:
       - hosts
       - collection
       - password
       - username
       - database
+      - operation
     type: object
     properties:
       hosts:
@@ -48,6 +61,22 @@ spec:
         title: MongoDB Database
         description: Sets the name of the MongoDB database to target.
         type: string
+      operation:
+        title: Operation
+        description: Which operation to execute on MongoDB findById, findOneByQuery, findAll, findDistinct, insert, save, update, remove, bulkWrite, aggregate, getDbStats, getColStats, count, command.
+        type: string
+        default: insert
+      writeConcern:
+        title: Write Concern
+        description: Configure the level of acknowledgment requested from MongoDB for write operations, possible values are ACKNOWLEDGED, W1, W2, W3, UNACKNOWLEDGED, JOURNALED, MAJORITY.
+        type: string
+      createCollection:
+        title: Collection
+        description: Create collection during initialisation if it doesn't exist.
+        type: boolean
+        default: false
+        x-descriptors:
+        - 'urn:alm:descriptor:com.tectonic.ui:checkbox'
   types:
     in:
       mediaType: application/json
@@ -62,12 +91,95 @@ spec:
     from:
       uri: kamelet:source
       steps:
+      - choice:
+          when:
+          - simple: "${header[criteria]}"
+            steps:
+            - set-header:
+                name: CamelMongoDbCriteria
+                simple: "${header[criteria]}"
+          - simple: "${header[ce-criteria]}"
+            steps:
+            - set-header:
+                name: CamelMongoDbCriteria
+                simple: "${header[ce-criteria]}"
+          - simple: "${header[num-skip]}"
+            steps:
+            - set-header:
+                name: CamelMongoDbNumToSkip
+                simple: "${header[num-skip]}"
+          - simple: "${header[ce-num-skip]}"
+            steps:
+            - set-header:
+                name: CamelMongoDbNumToSkip
+                simple: "${header[ce-num-skip]}"
+          - simple: "${header[limit]}"
+            steps:
+            - set-header:
+                name: CamelMongoDbLimit
+                simple: "${header[limit]}"
+          - simple: "${header[ce-limit]}"
+            steps:
+            - set-header:
+                name: CamelMongoDbLimit
+                simple: "${header[ce-limit]}"
+          - simple: "${header[batch-size]}"
+            steps:
+            - set-header:
+                name: CamelMongoDbBatchSize
+                simple: "${header[batch-size]}"
+          - simple: "${header[ce-batch-size]}"
+            steps:
+            - set-header:
+                name: CamelMongoDbBatchSize
+                simple: "${header[ce-batch-size]}"
+          - simple: "${header[sort-by]}"
+            steps:
+            - set-header:
+                name: CamelMongoDbSortBy
+                simple: "${header[sort-by]}"
+          - simple: "${header[ce-sort-by]}"
+            steps:
+            - set-header:
+                name: CamelMongoDbSortBy
+                simple: "${header[ce-sort-by]}"
+          - simple: "${header[multi-update]}"
+            steps:
+            - set-header:
+                name: CamelMongoDbMultiUpdate
+                simple: "${header[multi-update]}"
+          - simple: "${header[ce-multi-update]}"
+            steps:
+            - set-header:
+                name: CamelMongoDbMultiUpdate
+                simple: "${header[ce-multi-update]}"
+          - simple: "${header[db-upsert]}"
+            steps:
+            - set-header:
+                name: CamelMongoDbUpsert
+                simple: "${header[db-upsert]}"
+          - simple: "${header[ce-db-upsert]}"
+            steps:
+            - set-header:
+                name: CamelMongoDbUpsert
+                simple: "${header[ce-db-upsert]}"
+          - simple: "${header[collection-index]}"
+            steps:
+            - set-header:
+                name: CamelMongoDbCollectionIndex
+                simple: "${header[collection-index]}"
+          - simple: "${header[ce-collection-index]}"
+            steps:
+            - set-header:
+                name: CamelMongoDbCollectionIndex
+                simple: "${header[ce-collection-index]}"
       - to: 
           uri: "{{local-mongodb}}:test"
           parameters:
+            createCollection: "{{?createCollection}}"
             hosts: "{{hosts}}"
             collection: "{{collection}}"
             password: "{{password}}"
             username: "{{username}}"
             database: "{{database}}"
-            operation: "insert"
+            operation: "{{operation}}"