001/*
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.camel.management.mbean;
018
019import java.util.concurrent.BlockingQueue;
020import java.util.concurrent.ThreadPoolExecutor;
021import java.util.concurrent.TimeUnit;
022
023import org.apache.camel.CamelContext;
024import org.apache.camel.api.management.ManagedResource;
025import org.apache.camel.api.management.mbean.ManagedThreadsMBean;
026import org.apache.camel.model.ProcessorDefinition;
027import org.apache.camel.processor.ThreadsProcessor;
028
029@ManagedResource(description = "Managed Threads")
030public class ManagedThreads extends ManagedProcessor implements ManagedThreadsMBean {
031
032    public ManagedThreads(CamelContext context, ThreadsProcessor processor, ProcessorDefinition<?> definition) {
033        super(context, processor, definition);
034    }
035
036    @Override
037    public ThreadsProcessor getProcessor() {
038        return (ThreadsProcessor) super.getProcessor();
039    }
040
041    @Override
042    public Boolean isCallerRunsWhenRejected() {
043        if (getProcessor().getExecutorService() instanceof ThreadPoolExecutor) {
044            String name = getRejectedPolicy();
045            return "CallerRuns".equals(name);
046        } else {
047            return null;
048        }
049    }
050
051    @Override
052    public String getRejectedPolicy() {
053        if (getProcessor().getExecutorService() instanceof ThreadPoolExecutor threadPoolExecutor) {
054            return threadPoolExecutor.getRejectedExecutionHandler().toString();
055        } else {
056            return null;
057        }
058    }
059
060    @Override
061    public int getCorePoolSize() {
062        if (getProcessor().getExecutorService() instanceof ThreadPoolExecutor threadPoolExecutor) {
063            return threadPoolExecutor.getCorePoolSize();
064        } else {
065            return 0;
066        }
067    }
068
069    @Override
070    public int getPoolSize() {
071        if (getProcessor().getExecutorService() instanceof ThreadPoolExecutor threadPoolExecutor) {
072            return threadPoolExecutor.getPoolSize();
073        } else {
074            return 0;
075        }
076    }
077
078    @Override
079    public int getMaximumPoolSize() {
080        if (getProcessor().getExecutorService() instanceof ThreadPoolExecutor threadPoolExecutor) {
081            return threadPoolExecutor.getMaximumPoolSize();
082        } else {
083            return 0;
084        }
085    }
086
087    @Override
088    public int getLargestPoolSize() {
089        if (getProcessor().getExecutorService() instanceof ThreadPoolExecutor threadPoolExecutor) {
090            return threadPoolExecutor.getLargestPoolSize();
091        } else {
092            return 0;
093        }
094    }
095
096    @Override
097    public int getActiveCount() {
098        if (getProcessor().getExecutorService() instanceof ThreadPoolExecutor threadPoolExecutor) {
099            return threadPoolExecutor.getActiveCount();
100        } else {
101            return 0;
102        }
103    }
104
105    @Override
106    public long getTaskCount() {
107        if (getProcessor().getExecutorService() instanceof ThreadPoolExecutor threadPoolExecutor) {
108            return threadPoolExecutor.getTaskCount();
109        } else {
110            return 0;
111        }
112    }
113
114    @Override
115    public long getCompletedTaskCount() {
116        if (getProcessor().getExecutorService() instanceof ThreadPoolExecutor threadPoolExecutor) {
117            return threadPoolExecutor.getCompletedTaskCount();
118        } else {
119            return 0;
120        }
121    }
122
123    @Override
124    public long getTaskQueueSize() {
125        if (getProcessor().getExecutorService() instanceof ThreadPoolExecutor threadPoolExecutor) {
126            BlockingQueue<Runnable> queue = threadPoolExecutor.getQueue();
127            return queue != null ? queue.size() : 0;
128        } else {
129            return 0;
130        }
131    }
132
133    @Override
134    public long getKeepAliveTime() {
135        if (getProcessor().getExecutorService() instanceof ThreadPoolExecutor threadPoolExecutor) {
136            return threadPoolExecutor.getKeepAliveTime(TimeUnit.SECONDS);
137        } else {
138            return 0;
139        }
140    }
141
142    @Override
143    public boolean isAllowCoreThreadTimeout() {
144        if (getProcessor().getExecutorService() instanceof ThreadPoolExecutor threadPoolExecutor) {
145            return threadPoolExecutor.allowsCoreThreadTimeOut();
146        } else {
147            return false;
148        }
149    }
150
151}