private Pipeline(Builder builder) {
		// Copy the callback.
		mOnWorkUnitCompleteCallback = builder.mOnWorkUnitCompleteCallback;
		mOnFilterCreateCallback = builder.mOnFilterCreateCallback;
		
		// Create the filters.
		for (Class<? extends Filter> clazz : builder.mFilterClasses) {
			try {
				Filter filter = clazz.newInstance();
				
				if (mOnFilterCreateCallback != null) {
					mOnFilterCreateCallback.onCreate(filter);
				}
				
				mFilters.add(filter);
				
			} catch (InstantiationException e) {
				throw new IllegalArgumentException("Unable to instantiate " +
						"filter " + e.getClass().getName());
			} catch (IllegalAccessException e) {
				throw new IllegalArgumentException("Unable to instantiate " +
						"filter " + e.getClass().getName());
			}
		}
		
		// Set the input port to the source pipe.
		try {
			Filter filter = mFilters.get(0);
			filter.setInputPort(mSourcePipe);
		} catch (IndexOutOfBoundsException e) {
			// No filters!
			return;
		}
		
		// Connect the sink pipe to the sink filter.
		Filter filter = new SinkFilter();
		mFilters.add(filter);
		
		// Create pipes for each filter.
		for (int i = 0; i < mFilters.size(); ++i) {
			Pipe pipe = new Pipe();
			
			// Connect the output for the current filter to the pipe.
			Filter currentFilter = mFilters.get(i);
			currentFilter.setOutputPort(pipe);
			
			// If there exists a next filter, connect it to the pipe.
			if (i + 1 < mFilters.size()) {
				Filter nextFilter = mFilters.get(i + 1);
				nextFilter.setInputPort(pipe);
			}
		}
	}