You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@camel.apache.org by Chris Wolf <cw...@gmail.com> on 2013/03/25 21:08:23 UTC
Question about the implementation of routepolicy.quartz.ScheduledRoutePolicy
I am looking at code in
org.apache.camel.routepolicy.quartz.ScheduledRoutePolicy -
the code that starts/stops/resumes/suspends the route it is a policy for.
The question is why is it separately acting on the Consumer? I thought
if you call
CamelContext.suspendRoute(route) and/or CamelContext.resumeRoute(route),
that all the components in the route are suspended/resumed as well?
Secondly, why does this code assume there is a Consumer? What if the route has
a Producer?
I am not being critical - I just want to understand, generally how
route suspend/resume works
since I am implementing a similar RoutePolicy.
Thanks,
-Chris
protected void onJobExecute(Action action, Route route) throws Exception {
[...]
} else if (action == Action.SUSPEND) {
if (routeStatus == ServiceStatus.Started) {
stopConsumer(route.getConsumer());
} else {
LOG.warn("Route is not in a started state and cannot
be suspended. The current route state is {}", routeStatus);
}
} else if (action == Action.RESUME) {
if (routeStatus == ServiceStatus.Started) {
if (ServiceHelper.isSuspended(route.getConsumer())) {
startConsumer(route.getConsumer());
} else {
LOG.warn("The Consumer {} is not suspended and
cannot be resumed.", route.getConsumer());
}
[...]
Re: Question about the implementation of routepolicy.quartz.ScheduledRoutePolicy
Posted by Claus Ibsen <cl...@gmail.com>.
On Mon, Mar 25, 2013 at 9:08 PM, Chris Wolf <cw...@gmail.com> wrote:
> I am looking at code in
> org.apache.camel.routepolicy.quartz.ScheduledRoutePolicy -
> the code that starts/stops/resumes/suspends the route it is a policy for.
>
> The question is why is it separately acting on the Consumer? I thought
> if you call
> CamelContext.suspendRoute(route) and/or CamelContext.resumeRoute(route),
> that all the components in the route are suspended/resumed as well?
>
No a suspend/resume is a "light" action, that keeps the existing
resources "warm".
For example its used by a throttler route policy, to continuously
suspend/resume a route based on metrics.
You use case should use stop|start instead. Only use suspend/resume if
the duration is short, and you want to keep resources "warm".
> Secondly, why does this code assume there is a Consumer? What if the route has
> a Producer?
>
All routes starts from a consumer.
> I am not being critical - I just want to understand, generally how
> route suspend/resume works
> since I am implementing a similar RoutePolicy.
>
See the lifecycle documentation
http://camel.apache.org/lifecycle
> Thanks,
>
>
> -Chris
>
>
> protected void onJobExecute(Action action, Route route) throws Exception {
> [...]
> } else if (action == Action.SUSPEND) {
> if (routeStatus == ServiceStatus.Started) {
> stopConsumer(route.getConsumer());
> } else {
> LOG.warn("Route is not in a started state and cannot
> be suspended. The current route state is {}", routeStatus);
> }
> } else if (action == Action.RESUME) {
> if (routeStatus == ServiceStatus.Started) {
> if (ServiceHelper.isSuspended(route.getConsumer())) {
> startConsumer(route.getConsumer());
> } else {
> LOG.warn("The Consumer {} is not suspended and
> cannot be resumed.", route.getConsumer());
> }
>
> [...]
--
Claus Ibsen
-----------------
Red Hat, Inc.
FuseSource is now part of Red Hat
Email: cibsen@redhat.com
Web: http://fusesource.com
Twitter: davsclaus
Blog: http://davsclaus.com
Author of Camel in Action: http://www.manning.com/ibsen
Re: Question about the implementation of routepolicy.quartz.ScheduledRoutePolicy
Posted by Chris Wolf <cw...@gmail.com>.
As for the issue with the consumer appearing to continue to poll - I
think my cron resume/suspend expressions were overlapping.
Thanks,
Chris
On Tue, Mar 26, 2013 at 10:06 PM, Chris Wolf <cw...@gmail.com> wrote:
> Thanks for the clarification on that. The only problem is that, when
> using CronScheduledRoutePolicy
> on a route with an FTP endpoint configured as a consumer, it does NOT
> seem to prevent the FTP
> consumer to stop polling even after it has supposedly been suspended.
>
>
> 56:30,001 CronScheduledRoutePolicy DEBUG Suspended consumer
> FtpConsumer[ftp://localhost/download?filter=%23cpmdFileFilter&noop=true&password=******&username=adpt5]
> 56:35,100 FtpConsumer WARN Cannot connect/login
> to: ftp://adpt5@localhost:21. Will skip this poll.
> 56:35,100 FtpConsumer DEBUG Skipping poll as pre
> poll check returned false
>
>
> Any ideas?
>
> Thanks,
>
> Chris
>
> On Tue, Mar 26, 2013 at 9:11 PM, Raul Kripalani <ra...@evosent.com> wrote:
>> You don't want to stop all endpoints in the route because there may be
>> inflight Exchanges. Instead, you want to stop the inflow of new messages
>> whilst you keep processing any inflight exchanges.
>>
>> That said, it's safe to stop the entire route (including endpoints) once
>> all inflight exchanges are completed (or immediately if none are alive) –
>> as keeping endpoints started unnecessarily could be a waste of resources.
>> Please feel free to log a JIRA.
>>
>> On the other hand, all routes start with a consumer of some kind. Mind you,
>> it doesn't have to listen on an external protocol. Take for example the
>> direct, seda or timer components, which are "virtual" endpoints.
>>
>> Regards,
>>
>> *Raúl Kripalani*
>> Enterprise Architect, Open Source Integration specialist, Program
>> Manager | Apache
>> Camel Committer
>> http://about.me/raulkripalani | http://www.linkedin.com/in/raulkripalani
>> http://blog.raulkr.net | twitter: @raulvk
>>
>> On Mon, Mar 25, 2013 at 8:08 PM, Chris Wolf <cw...@gmail.com> wrote:
>>
>>> I am looking at code in
>>> org.apache.camel.routepolicy.quartz.ScheduledRoutePolicy -
>>> the code that starts/stops/resumes/suspends the route it is a policy for.
>>>
>>> The question is why is it separately acting on the Consumer? I thought
>>> if you call
>>> CamelContext.suspendRoute(route) and/or CamelContext.resumeRoute(route),
>>> that all the components in the route are suspended/resumed as well?
>>>
>>> Secondly, why does this code assume there is a Consumer? What if the
>>> route has
>>> a Producer?
>>>
>>> I am not being critical - I just want to understand, generally how
>>> route suspend/resume works
>>> since I am implementing a similar RoutePolicy.
>>>
>>> Thanks,
>>>
>>>
>>> -Chris
>>>
>>>
>>> protected void onJobExecute(Action action, Route route) throws Exception {
>>> [...]
>>> } else if (action == Action.SUSPEND) {
>>> if (routeStatus == ServiceStatus.Started) {
>>> stopConsumer(route.getConsumer());
>>> } else {
>>> LOG.warn("Route is not in a started state and cannot
>>> be suspended. The current route state is {}", routeStatus);
>>> }
>>> } else if (action == Action.RESUME) {
>>> if (routeStatus == ServiceStatus.Started) {
>>> if (ServiceHelper.isSuspended(route.getConsumer())) {
>>> startConsumer(route.getConsumer());
>>> } else {
>>> LOG.warn("The Consumer {} is not suspended and
>>> cannot be resumed.", route.getConsumer());
>>> }
>>>
>>> [...]
>>>
Re: Question about the implementation of routepolicy.quartz.ScheduledRoutePolicy
Posted by Claus Ibsen <cl...@gmail.com>.
On Wed, Mar 27, 2013 at 3:06 AM, Chris Wolf <cw...@gmail.com> wrote:
> Thanks for the clarification on that. The only problem is that, when
> using CronScheduledRoutePolicy
> on a route with an FTP endpoint configured as a consumer, it does NOT
> seem to prevent the FTP
> consumer to stop polling even after it has supposedly been suspended.
>
That is because you use pollEnrich on a FTP endpoint.
You need to stop instead of suspend, as stop will stop all the
resources of the route, and thus also pollEnrich.
>
> 56:30,001 CronScheduledRoutePolicy DEBUG Suspended consumer
> FtpConsumer[ftp://localhost/download?filter=%23cpmdFileFilter&noop=true&password=******&username=adpt5]
> 56:35,100 FtpConsumer WARN Cannot connect/login
> to: ftp://adpt5@localhost:21. Will skip this poll.
> 56:35,100 FtpConsumer DEBUG Skipping poll as pre
> poll check returned false
>
>
> Any ideas?
>
> Thanks,
>
> Chris
>
> On Tue, Mar 26, 2013 at 9:11 PM, Raul Kripalani <ra...@evosent.com> wrote:
>> You don't want to stop all endpoints in the route because there may be
>> inflight Exchanges. Instead, you want to stop the inflow of new messages
>> whilst you keep processing any inflight exchanges.
>>
>> That said, it's safe to stop the entire route (including endpoints) once
>> all inflight exchanges are completed (or immediately if none are alive) –
>> as keeping endpoints started unnecessarily could be a waste of resources.
>> Please feel free to log a JIRA.
>>
>> On the other hand, all routes start with a consumer of some kind. Mind you,
>> it doesn't have to listen on an external protocol. Take for example the
>> direct, seda or timer components, which are "virtual" endpoints.
>>
>> Regards,
>>
>> *Raúl Kripalani*
>> Enterprise Architect, Open Source Integration specialist, Program
>> Manager | Apache
>> Camel Committer
>> http://about.me/raulkripalani | http://www.linkedin.com/in/raulkripalani
>> http://blog.raulkr.net | twitter: @raulvk
>>
>> On Mon, Mar 25, 2013 at 8:08 PM, Chris Wolf <cw...@gmail.com> wrote:
>>
>>> I am looking at code in
>>> org.apache.camel.routepolicy.quartz.ScheduledRoutePolicy -
>>> the code that starts/stops/resumes/suspends the route it is a policy for.
>>>
>>> The question is why is it separately acting on the Consumer? I thought
>>> if you call
>>> CamelContext.suspendRoute(route) and/or CamelContext.resumeRoute(route),
>>> that all the components in the route are suspended/resumed as well?
>>>
>>> Secondly, why does this code assume there is a Consumer? What if the
>>> route has
>>> a Producer?
>>>
>>> I am not being critical - I just want to understand, generally how
>>> route suspend/resume works
>>> since I am implementing a similar RoutePolicy.
>>>
>>> Thanks,
>>>
>>>
>>> -Chris
>>>
>>>
>>> protected void onJobExecute(Action action, Route route) throws Exception {
>>> [...]
>>> } else if (action == Action.SUSPEND) {
>>> if (routeStatus == ServiceStatus.Started) {
>>> stopConsumer(route.getConsumer());
>>> } else {
>>> LOG.warn("Route is not in a started state and cannot
>>> be suspended. The current route state is {}", routeStatus);
>>> }
>>> } else if (action == Action.RESUME) {
>>> if (routeStatus == ServiceStatus.Started) {
>>> if (ServiceHelper.isSuspended(route.getConsumer())) {
>>> startConsumer(route.getConsumer());
>>> } else {
>>> LOG.warn("The Consumer {} is not suspended and
>>> cannot be resumed.", route.getConsumer());
>>> }
>>>
>>> [...]
>>>
--
Claus Ibsen
-----------------
Red Hat, Inc.
FuseSource is now part of Red Hat
Email: cibsen@redhat.com
Web: http://fusesource.com
Twitter: davsclaus
Blog: http://davsclaus.com
Author of Camel in Action: http://www.manning.com/ibsen
Re: Question about the implementation of routepolicy.quartz.ScheduledRoutePolicy
Posted by Chris Wolf <cw...@gmail.com>.
Thanks for the clarification on that. The only problem is that, when
using CronScheduledRoutePolicy
on a route with an FTP endpoint configured as a consumer, it does NOT
seem to prevent the FTP
consumer to stop polling even after it has supposedly been suspended.
56:30,001 CronScheduledRoutePolicy DEBUG Suspended consumer
FtpConsumer[ftp://localhost/download?filter=%23cpmdFileFilter&noop=true&password=******&username=adpt5]
56:35,100 FtpConsumer WARN Cannot connect/login
to: ftp://adpt5@localhost:21. Will skip this poll.
56:35,100 FtpConsumer DEBUG Skipping poll as pre
poll check returned false
Any ideas?
Thanks,
Chris
On Tue, Mar 26, 2013 at 9:11 PM, Raul Kripalani <ra...@evosent.com> wrote:
> You don't want to stop all endpoints in the route because there may be
> inflight Exchanges. Instead, you want to stop the inflow of new messages
> whilst you keep processing any inflight exchanges.
>
> That said, it's safe to stop the entire route (including endpoints) once
> all inflight exchanges are completed (or immediately if none are alive) –
> as keeping endpoints started unnecessarily could be a waste of resources.
> Please feel free to log a JIRA.
>
> On the other hand, all routes start with a consumer of some kind. Mind you,
> it doesn't have to listen on an external protocol. Take for example the
> direct, seda or timer components, which are "virtual" endpoints.
>
> Regards,
>
> *Raúl Kripalani*
> Enterprise Architect, Open Source Integration specialist, Program
> Manager | Apache
> Camel Committer
> http://about.me/raulkripalani | http://www.linkedin.com/in/raulkripalani
> http://blog.raulkr.net | twitter: @raulvk
>
> On Mon, Mar 25, 2013 at 8:08 PM, Chris Wolf <cw...@gmail.com> wrote:
>
>> I am looking at code in
>> org.apache.camel.routepolicy.quartz.ScheduledRoutePolicy -
>> the code that starts/stops/resumes/suspends the route it is a policy for.
>>
>> The question is why is it separately acting on the Consumer? I thought
>> if you call
>> CamelContext.suspendRoute(route) and/or CamelContext.resumeRoute(route),
>> that all the components in the route are suspended/resumed as well?
>>
>> Secondly, why does this code assume there is a Consumer? What if the
>> route has
>> a Producer?
>>
>> I am not being critical - I just want to understand, generally how
>> route suspend/resume works
>> since I am implementing a similar RoutePolicy.
>>
>> Thanks,
>>
>>
>> -Chris
>>
>>
>> protected void onJobExecute(Action action, Route route) throws Exception {
>> [...]
>> } else if (action == Action.SUSPEND) {
>> if (routeStatus == ServiceStatus.Started) {
>> stopConsumer(route.getConsumer());
>> } else {
>> LOG.warn("Route is not in a started state and cannot
>> be suspended. The current route state is {}", routeStatus);
>> }
>> } else if (action == Action.RESUME) {
>> if (routeStatus == ServiceStatus.Started) {
>> if (ServiceHelper.isSuspended(route.getConsumer())) {
>> startConsumer(route.getConsumer());
>> } else {
>> LOG.warn("The Consumer {} is not suspended and
>> cannot be resumed.", route.getConsumer());
>> }
>>
>> [...]
>>
Re: Question about the implementation of routepolicy.quartz.ScheduledRoutePolicy
Posted by Raul Kripalani <ra...@evosent.com>.
You don't want to stop all endpoints in the route because there may be
inflight Exchanges. Instead, you want to stop the inflow of new messages
whilst you keep processing any inflight exchanges.
That said, it's safe to stop the entire route (including endpoints) once
all inflight exchanges are completed (or immediately if none are alive) –
as keeping endpoints started unnecessarily could be a waste of resources.
Please feel free to log a JIRA.
On the other hand, all routes start with a consumer of some kind. Mind you,
it doesn't have to listen on an external protocol. Take for example the
direct, seda or timer components, which are "virtual" endpoints.
Regards,
*Raúl Kripalani*
Enterprise Architect, Open Source Integration specialist, Program
Manager | Apache
Camel Committer
http://about.me/raulkripalani | http://www.linkedin.com/in/raulkripalani
http://blog.raulkr.net | twitter: @raulvk
On Mon, Mar 25, 2013 at 8:08 PM, Chris Wolf <cw...@gmail.com> wrote:
> I am looking at code in
> org.apache.camel.routepolicy.quartz.ScheduledRoutePolicy -
> the code that starts/stops/resumes/suspends the route it is a policy for.
>
> The question is why is it separately acting on the Consumer? I thought
> if you call
> CamelContext.suspendRoute(route) and/or CamelContext.resumeRoute(route),
> that all the components in the route are suspended/resumed as well?
>
> Secondly, why does this code assume there is a Consumer? What if the
> route has
> a Producer?
>
> I am not being critical - I just want to understand, generally how
> route suspend/resume works
> since I am implementing a similar RoutePolicy.
>
> Thanks,
>
>
> -Chris
>
>
> protected void onJobExecute(Action action, Route route) throws Exception {
> [...]
> } else if (action == Action.SUSPEND) {
> if (routeStatus == ServiceStatus.Started) {
> stopConsumer(route.getConsumer());
> } else {
> LOG.warn("Route is not in a started state and cannot
> be suspended. The current route state is {}", routeStatus);
> }
> } else if (action == Action.RESUME) {
> if (routeStatus == ServiceStatus.Started) {
> if (ServiceHelper.isSuspended(route.getConsumer())) {
> startConsumer(route.getConsumer());
> } else {
> LOG.warn("The Consumer {} is not suspended and
> cannot be resumed.", route.getConsumer());
> }
>
> [...]
>