3 * Copyright (C) 2009-2016 SonarSource SA
4 * mailto:contact AT sonarsource DOT com
6 * This program is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU Lesser General Public
8 * License as published by the Free Software Foundation; either
9 * version 3 of the License, or (at your option) any later version.
11 * This program is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 * Lesser General Public License for more details.
16 * You should have received a copy of the GNU Lesser General Public License
17 * along with this program; if not, write to the Free Software Foundation,
18 * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
20 package org.sonar.server.computation.taskprocessor;
22 import com.google.common.base.Optional;
23 import org.sonar.api.utils.log.Logger;
24 import org.sonar.api.utils.log.Loggers;
25 import org.sonar.ce.log.CeLogging;
26 import org.sonar.ce.queue.CeTask;
27 import org.sonar.ce.queue.CeTaskResult;
28 import org.sonar.ce.taskprocessor.CeTaskProcessor;
29 import org.sonar.core.util.logs.Profiler;
30 import org.sonar.db.ce.CeActivityDto;
31 import org.sonar.server.computation.queue.InternalCeQueue;
33 import static java.lang.String.format;
35 public class CeWorkerCallableImpl implements CeWorkerCallable {
37 private static final Logger LOG = Loggers.get(CeWorkerCallableImpl.class);
39 private final InternalCeQueue queue;
40 private final CeLogging ceLogging;
41 private final CeTaskProcessorRepository taskProcessorRepository;
43 public CeWorkerCallableImpl(InternalCeQueue queue, CeLogging ceLogging, CeTaskProcessorRepository taskProcessorRepository) {
45 this.ceLogging = ceLogging;
46 this.taskProcessorRepository = taskProcessorRepository;
50 public Boolean call() throws Exception {
51 Optional<CeTask> ceTask = tryAndFindTaskToExecute();
52 if (!ceTask.isPresent()) {
57 executeTask(ceTask.get());
58 } catch (Exception e) {
59 LOG.error("An error occurred while managing task " + ceTask.get().getUuid(), e);
64 private Optional<CeTask> tryAndFindTaskToExecute() {
67 } catch (Exception e) {
68 LOG.error("Failed to pop the queue of analysis reports", e);
70 return Optional.absent();
73 private void executeTask(CeTask task) {
74 ceLogging.initForTask(task);
75 Profiler ceProfiler = startActivityProfiler(task);
77 CeActivityDto.Status status = CeActivityDto.Status.FAILED;
78 CeTaskResult taskResult = null;
79 Throwable error = null;
81 // TODO delegate the message to the related task processor, according to task type
82 Optional<CeTaskProcessor> taskProcessor = taskProcessorRepository.getForCeTask(task);
83 if (taskProcessor.isPresent()) {
84 taskResult = taskProcessor.get().process(task);
85 status = CeActivityDto.Status.SUCCESS;
87 LOG.error("No CeTaskProcessor is defined for task of type {}. Plugin configuration may have changed", task.getType());
88 status = CeActivityDto.Status.FAILED;
90 } catch (Throwable e) {
91 LOG.error(format("Failed to execute task %s", task.getUuid()), e);
94 queue.remove(task, status, taskResult, error);
95 stopActivityProfiler(ceProfiler, task, status);
96 ceLogging.clearForTask();
100 private static Profiler startActivityProfiler(CeTask task) {
101 Profiler profiler = Profiler.create(LOG);
102 addContext(profiler, task);
103 return profiler.startInfo("Execute task");
106 private static void stopActivityProfiler(Profiler profiler, CeTask task, CeActivityDto.Status status) {
107 addContext(profiler, task);
108 if (status == CeActivityDto.Status.FAILED) {
109 profiler.stopError("Executed task");
111 profiler.stopInfo("Executed task");
115 private static void addContext(Profiler profiler, CeTask task) {
118 .addContext("project", task.getComponentKey())
119 .addContext("type", task.getType())
120 .addContext("id", task.getUuid());
121 String submitterLogin = task.getSubmitterLogin();
122 if (submitterLogin != null) {
123 profiler.addContext("submitter", submitterLogin);