diff --git a/2021.11.16.txt b/2021.11.16.txt new file mode 100644 index 0000000..01761d5 --- /dev/null +++ b/2021.11.16.txt @@ -0,0 +1,73 @@ +package com.release11; + +import com.mysql.cj.jdbc.MysqlDataSource; +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.command.ActiveMQMapMessage; +import org.apache.camel.*; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.jms.JmsComponent; +import org.apache.camel.impl.DefaultCamelContext; +import org.apache.camel.support.SimpleRegistry; +import org.apache.log4j.BasicConfigurator; + +public class Main { + + public static void main(String[] args) throws Exception { + + //MyBuilder myBuilder = new MyBuilder(); + //myBuilder.doRandomMaterials(); + //myBuilder.doRandomPackage(); + + //to("log:?level=INFO&showHeaders=true&showBody=true") + + MysqlDataSource source = new MysqlDataSource(); + String jacek = "jdbc:mysql://10.101.111.19:3306/camel_db"; + source.setURL(jacek); + source.setUser("root"); + source.setPassword("admin"); + SimpleRegistry registry = new SimpleRegistry(); + registry.bind("source", MysqlDataSource.class, source); + CamelContext context = new DefaultCamelContext(registry); + BasicConfigurator.configure(); + ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://10.101.111.19:8088"); + connectionFactory.setUserName("admin"); + connectionFactory.setPassword("admin"); + context.addComponent("activemq", JmsComponent.jmsComponentAutoAcknowledge(connectionFactory)); + + DAO myFunctions = new DAO(); + + + + + + context.addRoutes(new RouteBuilder() { + @Override + public void configure() throws Exception { + + from("direct:start") + .setBody(constant("SELECT * FROM material LIMIT 10")) + .to("jdbc:source") + .split(body()) + .to("activemq:queue:material"); + + from("activemq:queue:material") + .split(body()) + //.bean(myFunctions, "getId") + .setHeader("id", simple("${body[id]}")) + .setBody(constant("SELECT * FROM package WHERE material_id = :?id")) + .to("jdbc:source?useHeadersAsParameters=true") + .to("activemq:queue:materialPackage"); + //.to("log:?level=INFO&showBody=true"); + + } + }); + + context.start(); + ProducerTemplate template = context.createProducerTemplate(); + template.sendBody("direct:start", null); + Thread.sleep(1000); + //template.setDefaultEndpointUri("activemq:queue:material"); + + context.stop(); + } +} diff --git a/src/main/java/com/release11/Main.java b/src/main/java/com/release11/Main.java index bff3989..963b31a 100644 --- a/src/main/java/com/release11/Main.java +++ b/src/main/java/com/release11/Main.java @@ -45,19 +45,20 @@ public class Main { public void configure() throws Exception { from("direct:start") - .setBody(constant("SELECT * FROM material LIMIT 10")) + .setBody(constant("SELECT * FROM material")) .to("jdbc:source") .split(body()) .to("activemq:queue:material"); + from("direct:getPackages") + .setHeader("material_id", simple("${body[id]}")) + .setBody(simple("SELECT * FROM package WHERE material_id = :?material_id")) + .to("jdbc:source?useHeadersAsParameters=true"); + from("activemq:queue:material") - .split(body()) - //.bean(myFunctions, "getId") - .setHeader("id", simple("${body[id]}")) - .setBody(constant("SELECT * FROM package WHERE material_id = :?id")) - .to("jdbc:source?useHeadersAsParameters=true") - .to("activemq:queue:materialPackage"); - //.to("log:?level=INFO&showBody=true"); + .enrich("direct:getPackages",new MyAggregator()) + .to("activemq:queue:materialPackage") + .to("log:?level=INFO&showBody=true"); } }); diff --git a/src/main/java/com/release11/MyAggregator.java b/src/main/java/com/release11/MyAggregator.java index f82b885..488d3f9 100644 --- a/src/main/java/com/release11/MyAggregator.java +++ b/src/main/java/com/release11/MyAggregator.java @@ -3,15 +3,26 @@ package com.release11; import org.apache.camel.AggregationStrategy; import org.apache.camel.Exchange; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.Map; + public class MyAggregator implements AggregationStrategy { @Override public Exchange aggregate(Exchange oldExchange, Exchange newExchange) { - String oldEx = oldExchange.getIn().getBody().toString(); + HashMap oldEx = (HashMap) oldExchange.getIn().getBody(); System.out.println(oldEx); - String newEx = newExchange.getIn().getBody().toString(); + ArrayList newEx = (ArrayList) newExchange.getIn().getBody(); System.out.println(newEx); - return null; + + oldEx.put("packages", newEx); + Exchange result = oldExchange; + result.getIn().setBody(oldEx); + //result.getIn().setBody(body); + System.out.println(result.getIn().getBody()); + + return result; } } diff --git a/target/classes/com/release11/Main$1.class b/target/classes/com/release11/Main$1.class index 06d02fc..0941cc6 100644 Binary files a/target/classes/com/release11/Main$1.class and b/target/classes/com/release11/Main$1.class differ diff --git a/target/classes/com/release11/Main.class b/target/classes/com/release11/Main.class index 0e35f97..99c1953 100644 Binary files a/target/classes/com/release11/Main.class and b/target/classes/com/release11/Main.class differ diff --git a/target/classes/com/release11/MyAggregator.class b/target/classes/com/release11/MyAggregator.class new file mode 100644 index 0000000..0897f6a Binary files /dev/null and b/target/classes/com/release11/MyAggregator.class differ