You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@camel.apache.org by suman <pr...@gmail.com> on 2012/01/19 17:30:43 UTC
Converting message body after GroupedExchangeAggregationStrategy
Hello All,
This is my route
from("direct:getstreaminfo").startupOrder(2)
.recipientList().method(MessageRouter.class,
"routeTo").parallelProcessing().aggregationStrategy(new
GroupedExchangeAggregationStrategy()).convertBodyTo(String.class)
.process(eventStreamedProcessor).to("log:foo");
Iam trying to get the list of exchanges in processor using
List<Exchange> exchangeList =
exchange.getProperty(Exchange.GROUPED_EXCHANGE, List.class);
But my problem is within each of the exchanges,the body is of type
CachedOutputStream.WrappedInputStream which i cant access.
So all i want is within each of the exchange present in the list,i want to
body to be of type String.Is it possible?
Thanks
Suman
--
View this message in context: http://camel.465427.n5.nabble.com/Converting-message-body-after-GroupedExchangeAggregationStrategy-tp5158191p5158191.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Converting message body after
GroupedExchangeAggregationStrategy
Posted by suman <pr...@gmail.com>.
Hello,
I tried putting implementation inside custom AggregationStrategy and it
seems to be working.I dont know why i was getting empty string when i did
debug this newexchange object in intellij.
Thanks for your help
public class MyAggregationStrategy implements AggregationStrategy {
@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
List<Exchange> list;
Exchange answer = oldExchange;
if (oldExchange == null) {
answer = new DefaultExchange(newExchange);
list = new ArrayList<Exchange>();
answer.setProperty(Exchange.GROUPED_EXCHANGE, list);
} else {
list = oldExchange.getProperty(Exchange.GROUPED_EXCHANGE,
List.class);
}
Message message = newExchange.getIn();
message.setBody(newExchange.getIn().getBody(String.class));
if (newExchange != null) {
list.add(newExchange);
}
return answer;
}
}
--
View this message in context: http://camel.465427.n5.nabble.com/Converting-message-body-after-GroupedExchangeAggregationStrategy-tp5158191p5160884.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Converting message body after
GroupedExchangeAggregationStrategy
Posted by suman <pr...@gmail.com>.
Ans there is no implementation side that aggregationstrategy because i just
put a breakpoint inside that class and manually evaluate newexchange to see
what it returns..
--
View this message in context: http://camel.465427.n5.nabble.com/Converting-message-body-after-GroupedExchangeAggregationStrategy-tp5158191p5160757.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Converting message body after
GroupedExchangeAggregationStrategy
Posted by suman <pr...@gmail.com>.
Sorry for confusion..I just debugged inside that
GroupedExchangeAggregationStrategy to see what newExchange value
has..Anyways in order to avoid confusion i created a dummy one..But still
the problem persists...
public class MyAggregationStrategy implements AggregationStrategy {
@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
return null; //To change body of implemented methods use File |
Settings | File Templates.
}
}
and this is my route
from("direct:getlivestreaminfo").startupOrder(2).convertBodyTo(String.class)
.recipientList().method(messageRouter,
"routeTo").parallelProcessing().aggregationStrategy(new
MyAggregationStrategy())
.process(eventStreamedProcessor).to("log:foo");
--
View this message in context: http://camel.465427.n5.nabble.com/Converting-message-body-after-GroupedExchangeAggregationStrategy-tp5158191p5160748.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Converting message body after GroupedExchangeAggregationStrategy
Posted by Claus Ibsen <cl...@gmail.com>.
On Fri, Jan 20, 2012 at 4:22 PM, suman <pr...@gmail.com> wrote:
> No..The body inside that exchange is of type
> org.apache.camel.converter.stream.CachedOutputStream$WrappedInputStream@4b8b7245
> And when i just debug it inside GroupedExchangeAggregationStrategy and
> evaluate newexchange using
> newExchange.getIn().getBody(String.class),its returning empty string.("")
> So either the conversion isnt working or the stream is closed at that point?
> This is my route
>
As I told you do NOT use the GroupedExchangeAggregationStrategy.
Build your custom strategy and convert the payload to a String.
> from("direct:getlivestreaminfo").startupOrder(2).convertBodyTo(String.class)
> .recipientList().method(messageRouter,
> "routeTo").parallelProcessing().aggregationStrategy(new
> GroupedExchangeAggregationStrategy())
> .process(eventStreamedAggregationProcessor).to("log:foo");
>
>
> --
> View this message in context: http://camel.465427.n5.nabble.com/Converting-message-body-after-GroupedExchangeAggregationStrategy-tp5158191p5160728.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
--
Claus Ibsen
-----------------
FuseSource
Email: cibsen@fusesource.com
Web: http://fusesource.com
Twitter: davsclaus, fusenews
Blog: http://davsclaus.blogspot.com/
Author of Camel in Action: http://www.manning.com/ibsen/
Re: Converting message body after
GroupedExchangeAggregationStrategy
Posted by suman <pr...@gmail.com>.
No..The body inside that exchange is of type
org.apache.camel.converter.stream.CachedOutputStream$WrappedInputStream@4b8b7245
And when i just debug it inside GroupedExchangeAggregationStrategy and
evaluate newexchange using
newExchange.getIn().getBody(String.class),its returning empty string.("")
So either the conversion isnt working or the stream is closed at that point?
This is my route
from("direct:getlivestreaminfo").startupOrder(2).convertBodyTo(String.class)
.recipientList().method(messageRouter,
"routeTo").parallelProcessing().aggregationStrategy(new
GroupedExchangeAggregationStrategy())
.process(eventStreamedAggregationProcessor).to("log:foo");
--
View this message in context: http://camel.465427.n5.nabble.com/Converting-message-body-after-GroupedExchangeAggregationStrategy-tp5158191p5160728.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Converting message body after GroupedExchangeAggregationStrategy
Posted by Claus Ibsen <cl...@gmail.com>.
On Fri, Jan 20, 2012 at 4:09 PM, suman <pr...@gmail.com> wrote:
> I tried this but no luck...In my receipient list,the list of recipients are a
> comma separated HTTP URLs..
> So what iam trying to do is make a parallel request to bunch of HTTP urls
> and than try to process them in the
> processor(eventStreamedAggregationProcessor).Inside that processor, i
> retrieve list of exchanges but the body in each of them are of type
> org.apache.camel.converter.stream.CachedOutputStream$WrappedInputStream.
> And sometimes i can see that the stream is closed as well.
>
>
Ah implement your own AggregationStrategy, instead of the
GroupedExchangeAggregationStrategy, and get the body as a String and
store the bodies.
String body = newExchange.getIn().getBody(String.class);
> from("direct:getlivestreaminfo").startupOrder(2).convertBodyTo(String.class)
> .recipientList().method(messageRouter,
> "routeTo").parallelProcessing().aggregationStrategy(new
> GroupedExchangeAggregationStrategy())
> .process(eventStreamedAggregationProcessor).to("log:foo");
>
> --
> View this message in context: http://camel.465427.n5.nabble.com/Converting-message-body-after-GroupedExchangeAggregationStrategy-tp5158191p5160681.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
--
Claus Ibsen
-----------------
FuseSource
Email: cibsen@fusesource.com
Web: http://fusesource.com
Twitter: davsclaus, fusenews
Blog: http://davsclaus.blogspot.com/
Author of Camel in Action: http://www.manning.com/ibsen/
Re: Converting message body after
GroupedExchangeAggregationStrategy
Posted by suman <pr...@gmail.com>.
I tried this but no luck...In my receipient list,the list of recipients are a
comma separated HTTP URLs..
So what iam trying to do is make a parallel request to bunch of HTTP urls
and than try to process them in the
processor(eventStreamedAggregationProcessor).Inside that processor, i
retrieve list of exchanges but the body in each of them are of type
org.apache.camel.converter.stream.CachedOutputStream$WrappedInputStream.
And sometimes i can see that the stream is closed as well.
from("direct:getlivestreaminfo").startupOrder(2).convertBodyTo(String.class)
.recipientList().method(messageRouter,
"routeTo").parallelProcessing().aggregationStrategy(new
GroupedExchangeAggregationStrategy())
.process(eventStreamedAggregationProcessor).to("log:foo");
--
View this message in context: http://camel.465427.n5.nabble.com/Converting-message-body-after-GroupedExchangeAggregationStrategy-tp5158191p5160681.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Converting message body after GroupedExchangeAggregationStrategy
Posted by Claus Ibsen <cl...@gmail.com>.
Add the convertBodtTo before the recipient list.
On Fri, Jan 20, 2012 at 3:58 PM, suman <pr...@gmail.com> wrote:
> Thanks.
> I have tried it but the dsl doesn't compile(added the one in bold)
>
> from("direct:getlivestreaminfo").startupOrder(2)
> .recipientList().method(messageRouter,
> "routeTo").parallelProcessing().*convertBodyTo(String.class)*.aggregationStrategy(new
> GroupedExchangeAggregationStrategy())
> .process(eventStreamedAggregationStrategy).to("log:foo");
>
>
> --
> View this message in context: http://camel.465427.n5.nabble.com/Converting-message-body-after-GroupedExchangeAggregationStrategy-tp5158191p5160638.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
--
Claus Ibsen
-----------------
FuseSource
Email: cibsen@fusesource.com
Web: http://fusesource.com
Twitter: davsclaus, fusenews
Blog: http://davsclaus.blogspot.com/
Author of Camel in Action: http://www.manning.com/ibsen/
Re: Converting message body after
GroupedExchangeAggregationStrategy
Posted by suman <pr...@gmail.com>.
Thanks.
I have tried it but the dsl doesn't compile(added the one in bold)
from("direct:getlivestreaminfo").startupOrder(2)
.recipientList().method(messageRouter,
"routeTo").parallelProcessing().*convertBodyTo(String.class)*.aggregationStrategy(new
GroupedExchangeAggregationStrategy())
.process(eventStreamedAggregationStrategy).to("log:foo");
--
View this message in context: http://camel.465427.n5.nabble.com/Converting-message-body-after-GroupedExchangeAggregationStrategy-tp5158191p5160638.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Converting message body after GroupedExchangeAggregationStrategy
Posted by Claus Ibsen <cl...@gmail.com>.
On Thu, Jan 19, 2012 at 5:30 PM, suman <pr...@gmail.com> wrote:
> Hello All,
> This is my route
> from("direct:getstreaminfo").startupOrder(2)
> .recipientList().method(MessageRouter.class,
> "routeTo").parallelProcessing().aggregationStrategy(new
> GroupedExchangeAggregationStrategy()).convertBodyTo(String.class)
> .process(eventStreamedProcessor).to("log:foo");
>
> Iam trying to get the list of exchanges in processor using
> List<Exchange> exchangeList =
> exchange.getProperty(Exchange.GROUPED_EXCHANGE, List.class);
> But my problem is within each of the exchanges,the body is of type
> CachedOutputStream.WrappedInputStream which i cant access.
> So all i want is within each of the exchange present in the list,i want to
> body to be of type String.Is it possible?
>
Either convert the body to a String *before* sending to the aggregator.
Or convert the body to a String after.
I recommend to do it before as the stream may be closed on later
access, if it takes
too long time to access it afterwards.
> Thanks
> Suman
>
> --
> View this message in context: http://camel.465427.n5.nabble.com/Converting-message-body-after-GroupedExchangeAggregationStrategy-tp5158191p5158191.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
--
Claus Ibsen
-----------------
FuseSource
Email: cibsen@fusesource.com
Web: http://fusesource.com
Twitter: davsclaus, fusenews
Blog: http://davsclaus.blogspot.com/
Author of Camel in Action: http://www.manning.com/ibsen/