Creates a writer for each output. Creates an OutputCollector which forwards its input to all writers.
The output collector applies the configured shipping strategy.
555448
@SuppressWarnings("unchecked")
	public static <T> Collector<T> initOutputs(AbstractInvokable containingTask, ClassLoader cl, TaskConfig config,
										List<ChainedDriver<?, ?>> chainedTasksTarget,
										List<RecordWriter<?>> eventualOutputs,
										ExecutionConfig executionConfig,
										Map<String, Accumulator<?,?>> accumulatorMap)
	throws Exception
	{
		final int numOutputs = config.getNumOutputs();

		// check whether we got any chained tasks
		final int numChained = config.getNumberOfChainedStubs();
		if (numChained > 0) {
			// got chained stubs. that means that this one may only have a single forward connection
			if (numOutputs != 1 || config.getOutputShipStrategy(0) != ShipStrategyType.FORWARD) {
				throw new RuntimeException("Plan Generation Bug: Found a chained stub that is not connected via an only forward connection.");
			}

			// instantiate each task
			@SuppressWarnings("rawtypes")
			Collector previous = null;
			for (int i = numChained - 1; i >= 0; --i)
			{
				// get the task first
				final ChainedDriver<?, ?> ct;
				try {
					Class<? extends ChainedDriver<?, ?>> ctc = config.getChainedTask(i);
					ct = ctc.newInstance();
				}
				catch (Exception ex) {
					throw new RuntimeException("Could not instantiate chained task driver.", ex);
				}

				// get the configuration for the task
				final TaskConfig chainedStubConf = config.getChainedStubConfig(i);
				final String taskName = config.getChainedTaskName(i);

				if (i == numChained - 1) {
					// last in chain, instantiate the output collector for this task
					previous = getOutputCollector(containingTask, chainedStubConf, cl, eventualOutputs, 0, chainedStubConf.getNumOutputs());
				}

				ct.setup(chainedStubConf, taskName, previous, containingTask, cl, executionConfig, accumulatorMap);
				chainedTasksTarget.add(0, ct);

				if (i == numChained - 1) {
					ct.getIOMetrics().reuseOutputMetricsForTask();
				}

				previous = ct;
			}
			// the collector of the first in the chain is the collector for the task
			return (Collector<T>) previous;
		}
		// else

		// instantiate the output collector the default way from this configuration
		return getOutputCollector(containingTask , config, cl, eventualOutputs, 0, numOutputs);
	}
1365440
def writers(self):
        
        return list((fd, data) for fd, (data, gen) in self._wfds.items())
996998
public static StoreWriter createWriter(OutputStream stream, Configuration config) {
    return StoreImpl.createWriter(stream, config);
  }
812621
void prepare(Map stormConf, TopologyContext context, OutputCollector collector, State state) {
        init(context, collector);
        this.collector = new AckTrackingOutputCollector(collector);
        bolt.prepare(stormConf, context, this.collector);
        this.state = state;
    }
1080168
def addWriter(self, selectable):
        """"""
        try:
            self._writes[selectable].resume()
        except KeyError:
            self._writes[selectable] = g = Stream.spawn(self, selectable, 'doWrite')
            self.addToGreenletPool(g)
1206193
@SuppressWarnings({"unchecked"})
  public OutputCollector getCollector(String namedOutput, Reporter reporter)
    throws IOException {
    return getCollector(namedOutput, null, reporter);
  }
801418
public static void writeCollection(XMLOutput xmlOutput, Collection<? extends XMLWriteable> collection) throws IOException {
        for (XMLWriteable obj : collection) {
            obj.writeXML(xmlOutput);
        }
    }
1328538
public void generateOutput(Writer outputWriter) throws IOException {
        String s = generateOutput();
        outputWriter.write(s);
    }
809187
private void writeArray(SerIterator itemIterator) throws IOException {
        output.writeArrayStart();
        while (itemIterator.hasNext()) {
            itemIterator.next();
            output.writeArrayItemStart();
            writeObject(itemIterator.valueType(), itemIterator.value(), itemIterator);
        }
        output.writeArrayEnd();
    }
638957
private static BufferedWriter bufferOutput( Writer writer )
    {
        BufferedWriter out;
        if( writer instanceof BufferedWriter )
        {
            out = (BufferedWriter) writer;
        }
        else
        {
            out = new BufferedWriter( writer );
        }
        return out;
    }
776897
def setCollectors(self, collectors):
        
        self.__collectors = {}
        for name, collector in collectors.items():
            self.__collectors[name] = collector
            collector.setSchema(self)
85360
protected void writeCollectionValue(Collection<?> v) throws IOException
    {
        _generator.writeStartArray();
        for (Object ob : v) {
            writeValue(ob);
        }
        _generator.writeEndArray();
    }
736693
def with_output
      super do |out|
        out.puts header
        @queued_write = []
        yield out
        @queued_write.sort.each do |line|
          out.puts(line)
        end
      end
    end
391243
def generate_output(inputs, output)
      inputs.each do |input|
        output.write process_filters(input)
      end
    end
1224593
@Override
	protected void initialize() throws Exception {
		super.initialize();

		// sanity check: the tail has to update either the workset or the solution set
		if (!isWorksetUpdate && !isSolutionSetUpdate) {
			throw new RuntimeException("The iteration tail doesn't update workset or the solution set.");
		}

		// set the last output collector of this task to reflect the iteration tail state update:
		// a) workset update,
		// b) solution set update, or
		// c) merged workset and solution set update

		Collector<OT> outputCollector = null;
		if (isWorksetUpdate) {
			outputCollector = createWorksetUpdateOutputCollector();

			// we need the WorksetUpdateOutputCollector separately to count the collected elements
			if (isWorksetIteration) {
				worksetUpdateOutputCollector = (WorksetUpdateOutputCollector<OT>) outputCollector;
			}
		}

		if (isSolutionSetUpdate) {
			if (isWorksetIteration) {
				outputCollector = createSolutionSetUpdateOutputCollector(outputCollector);
			}
			// Bulk iteration with termination criterion
			else {
				outputCollector = new Collector<OT>() {
					@Override
					public void collect(OT record) {}

					@Override
					public void close() {}
				};
			}

			if (!isWorksetUpdate) {
				solutionSetUpdateBarrier = SolutionSetUpdateBarrierBroker.instance().get(brokerKey());
			}
		}

		setLastOutputCollector(outputCollector);
	}
888594
private static Map<WriterKind, PrintWriter> initWriters(PrintWriter out, PrintWriter err) {
        Map<WriterKind, PrintWriter> writers = new EnumMap<>(WriterKind.class);
        writers.put(WriterKind.ERROR, err);
        writers.put(WriterKind.WARNING, err);
        writers.put(WriterKind.NOTICE, err);

        writers.put(WriterKind.STDOUT, out);
        writers.put(WriterKind.STDERR, err);

        return writers;
    }
485007
public <T> Collector<T> create(long id, Collection<Address> backupOwners, int topologyId) {
      if (backupOwners.isEmpty()) {
         return new PrimaryOwnerOnlyCollector<>();
      }
      SingleKeyCollector<T> collector = new SingleKeyCollector<>(id, backupOwners, topologyId);
      BaseAckTarget prev = collectorMap.put(id, collector);
      //is it possible the have a previous collector when the topology changes after the first collector is created
      //in that case, the previous collector must have a lower topology id
      assert prev == null || prev.topologyId < topologyId : format("replaced old collector '%s' by '%s'", prev, collector);
      if (trace) {
         log.tracef("Created new collector for %s. BackupOwners=%s", id, backupOwners);
      }
      return collector;
   }