Utility Classes and Base Components

/**
 * Abstract base class for spouts that read line-based text data
 */
public abstract class AbstractLineSpout implements IRichSpout {
    public static final String ATTRIBUTE_LINE = "line";
    
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector);
    public void close();
    public void activate();
    public void deactivate();
    public void ack(Object msgId);
    public void fail(Object msgId);
    public void declareOutputFields(OutputFieldsDeclarer declarer);
    public Map<String, Object> getComponentConfiguration();
    
    // Implement in subclasses
    public abstract void nextTuple();
}

/**
 * Abstract base class for bolt sinks that write processed data
 */
public abstract class AbstractBoltSink implements IRichBolt {
    public AbstractBoltSink(OutputFormatter formatter);
    
    public final void prepare(Map stormConf, TopologyContext context, OutputCollector collector);
    public final void execute(Tuple input);
    public void cleanup();
    public final void declareOutputFields(OutputFieldsDeclarer declarer);
    public Map<String, Object> getComponentConfiguration();
    
    // Implement in subclasses
    protected abstract void prepareSimple(Map<?, ?> stormConf, TopologyContext context);
    protected abstract void writeExternal(String line);
}

/**
 * Spout for reading data from in-memory arrays
 * @param <T> Type of data elements
 */
public class InMemorySpout<T> extends AbstractLineSpout {
    /**
     * Create spout with data source array
     * @param source Array of data elements to emit
     */
    public InMemorySpout(T[] source);
    
    public void nextTuple();
}

/**
 * Memory spout that terminates when data is exhausted
 */
public class FiniteInMemorySpout extends InMemorySpout<String> implements FiniteSpout {
    /**
     * Create finite spout with string data source
     * @param source Array of strings to emit
     */
    public FiniteInMemorySpout(String[] source);
    
    /**
     * Check if spout has reached end of data
     * @return true if no more data available
     */
    public boolean reachedEnd();
}

/**
 * Spout for reading lines from local files
 */
public class FileSpout extends AbstractLineSpout {
    public static final String INPUT_FILE_PATH = "input.path";
    
    /**
     * Create file spout with path from configuration
     */
    public FileSpout();
    
    /**
     * Create file spout with specific path
     * @param path Path to input file
     */
    public FileSpout(String path);
    
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector);
    public void close();
    public void nextTuple();
}

/**
 * File spout that terminates at end of file
 */
public class FiniteFileSpout extends FileSpout implements FiniteSpout {
    public FiniteFileSpout();
    public FiniteFileSpout(String path);
    
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector);
    public void nextTuple();
    public boolean reachedEnd();
}

/**
 * Sink that prints tuples to standard output
 */
public class BoltPrintSink extends AbstractBoltSink {
    /**
     * Create print sink with formatter
     * @param formatter Formatter for tuple output
     */
    public BoltPrintSink(OutputFormatter formatter);
    
    protected void prepareSimple(Map stormConf, TopologyContext context);
    protected void writeExternal(String line);
}

/**
 * Sink that writes tuples to files
 */
public class BoltFileSink extends AbstractBoltSink {
    /**
     * Create file sink with path and default formatter
     * @param path Output file path
     */
    public BoltFileSink(String path);
    
    /**
     * Create file sink with path and custom formatter
     * @param path Output file path
     * @param formatter Formatter for tuple output
     */
    public BoltFileSink(String path, OutputFormatter formatter);
    
    protected void prepareSimple(Map stormConf, TopologyContext context);
    protected void writeExternal(String line);
    public void cleanup();
}

/**
 * Interface for custom tuple formatting
 */
public interface OutputFormatter extends Serializable {
    /**
     * Format tuple for output
     * @param input Tuple to format
     * @return Formatted string representation
     */
    String format(Tuple input);
}

/**
 * Formats single-field tuples as strings
 */
public class SimpleOutputFormatter implements OutputFormatter {
    /**
     * Format single-field tuple as string
     * @param input Tuple with single field
     * @return String representation of first field
     */
    public String format(Tuple input);
}

/**
 * Formats multi-field tuples in (field1,field2,...) format
 */
public class TupleOutputFormatter implements OutputFormatter {
    /**
     * Format multi-field tuple in parentheses format
     * @param input Tuple with multiple fields
     * @return String in format (field1,field2,...)
     */
    public String format(Tuple input);
}

import org.apache.flink.storm.util.InMemorySpout;

// Create spout with string data
String[] data = {"Hello", "World", "Storm", "Flink"};
InMemorySpout<String> spout = new InMemorySpout<>(data);

// Create finite spout that stops automatically
FiniteInMemorySpout finiteSpout = new FiniteInMemorySpout(data);

import org.apache.flink.storm.util.FileSpout;
import org.apache.flink.storm.util.FiniteFileSpout;

// Read from specific file
FileSpout fileSpout = new FileSpout("/path/to/input.txt");

// Read file and stop at end
FiniteFileSpout finiteFileSpout = new FiniteFileSpout("/path/to/input.txt");

import org.apache.flink.storm.util.*;

// Print to console with simple formatting
OutputFormatter simpleFormatter = new SimpleOutputFormatter();
BoltPrintSink printSink = new BoltPrintSink(simpleFormatter);

// Write to file with tuple formatting
OutputFormatter tupleFormatter = new TupleOutputFormatter();
BoltFileSink fileSink = new BoltFileSink("/path/to/output.txt", tupleFormatter);

import org.apache.flink.storm.util.AbstractLineSpout;

public class CustomSpout extends AbstractLineSpout {
    private String[] data;
    private int index = 0;
    
    public CustomSpout(String[] data) {
        this.data = data;
    }
    
    @Override
    public void nextTuple() {
        if (index < data.length) {
            collector.emit(new Values(data[index++]));
        }
    }
}