Integration with Kaoto

 
                //Start with a timer that executes the operation every 10 seconds
                from("timer:java?period=10000")
                      //Access the CSV file which can be on an api or storage device
                      .to("{{source.csv}}")
                      //unmarshal and split the workflow per row 
                      .unmarshal().split(body()).streaming()
                      //process each row
                      //the process is defined in a separated file
                      .process(processCsv)
                      //push the results to the following API
                      .to("https://nominatim.openstreetmap.org/reverse")
                      //the returned XML is also processed
                      .unmarshal().jacksonxml()
                      //again, the processor is defined in a separated file
                      .process(processXML)
                      //Prepare an SQL query based on the result of the process
                      .setBody().simple("SELECT info FROM descriptions WHERE id like '${exchangeProperty.pollutant}'")
                      //send the SQL query to the database
                      .to("jdbc:postgresBean?readSize=1")
                      //collect the result and process it
                      .process(processDB)
                      //reunite the parallel streams that started on the csv processing
                      .aggregate(constant(true), aggregationStrategy)
                      .completionSize(5)
                      //given the list of outputs per row in csv, process it
                      .process(buildGeoJSON)
                      //store the result in another database
                      .to("mongodb:mongoBean?database=example&collection=mySpatialObjects&operation=insert")