/*

* Copyright 2012 LinkedIn, Inc

*

* Licensed under the Apache License, Version 2.0 (the "License"); you may not

* use this file except in compliance with the License. You may obtain a copy of

* the License at

*

* http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT

* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the

* License for the specific language governing permissions and limitations under

* the License.

*/

package com.linkedin.parseq;

import com.linkedin.parseq.internal.InternalUtil;

import com.linkedin.parseq.internal.SystemHiddenTask;

import com.linkedin.parseq.promise.Promise;

import com.linkedin.parseq.promise.PromiseListener;

import com.linkedin.parseq.promise.Promises;

import com.linkedin.parseq.promise.SettablePromise;

import com.linkedin.parseq.trace.ResultType;

import com.linkedin.parseq.trace.ShallowTrace;

import com.linkedin.parseq.trace.ShallowTraceBuilder;

import java.util.ArrayList;

import java.util.Collections;

import java.util.List;

/**

* A {@link Task} that will run all of the constructor-supplied tasks in parallel.

* <p/>

* Use {@link Tasks#par(Task[])} or {@link Tasks#par(Iterable)} to create an

* instance of this class.

*

* @author Chris Pettitt (cpettitt@linkedin.com)

* @author Chi Chan (ckchan@linkedin.com)

*/

/* package private */ class ParTaskImpl<T> extends SystemHiddenTask<List<T>> implements ParTask<T>

{

private final List<Task<T>> _tasks;

public ParTaskImpl(final String name, final Iterable<? extends Task<? extends T>> tasks)

{

super(name);

List<Task<T>> taskList = new ArrayList<Task<T>>();

for(Task<? extends T> task : tasks)

{

// Safe to coerce Task<? extends T> to Task<T>

@SuppressWarnings("unchecked")

final Task<T> coercedTask = (Task<T>)task;

taskList.add(coercedTask);

}

if (taskList.isEmpty())

{

throw new IllegalArgumentException("No tasks to parallelize!");

}

_tasks = Collections.unmodifiableList(taskList);

}

@Override

protected Promise<List<T>> run(final Context context) throws Exception

{

final SettablePromise<List<T>> result = Promises.settable();

final PromiseListener<?> listener = new PromiseListener<Object>()

{

@Override

public void onResolved(Promise<Object> resolvedPromise)

{

boolean allEarlyFinish = true;

final List<T> taskResult = new ArrayList<T>();

final List<Throwable> errors = new ArrayList<Throwable>();

for (Task<? extends T> task : _tasks)

{

if (task.isFailed())

{

if (allEarlyFinish && ResultType.fromTask(task) != ResultType.EARLY_FINISH)

{

allEarlyFinish = false;

}

errors.add(task.getError());

}

else

{

taskResult.add(task.get());

}

}

if (!errors.isEmpty())

{

result.fail(allEarlyFinish ? errors.get(0) : new MultiException("Multiple errors in 'ParTask' task.", errors));

}

else

{

result.done(taskResult);

}

}

};

InternalUtil.after(listener, _tasks.toArray(new Task<?>[_tasks.size()]));

for (Task<?> task : _tasks)

{

context.run(task);

}

return result;

}

@Override

public List<Task<T>> getTasks()

{

return _tasks;

}

@Override

public List<T> getSuccessful()

{

if(!this.isFailed())

{

return this.get();

}

final List<T> taskResult = new ArrayList<T>();

for (Task<? extends T> task : _tasks)

{

if (!task.isFailed())

{

taskResult.add(task.get());

}

}

return taskResult;

}

}