Java 类com.amazonaws.services.kinesis.connectors.interfaces.IEmitter 实例源码

public KinesisConnectorRecordProcessor(IBuffer<T> buffer,
        IFilter<T> filter,
        IEmitter<U> emitter,
        ITransformerBase<T, U> transformer,
        KinesisConnectorConfiguration configuration) {
    if (buffer == null || filter == null || emitter == null || transformer == null) {
        throw new IllegalArgumentException("buffer, filter, emitter, and transformer must not be null");
    }
    this.buffer = buffer;
    this.filter = filter;
    this.emitter = emitter;
    this.transformer = transformer;
    // Limit must be greater than zero
    if (configuration.RETRY_LIMIT <= 0) {
        retryLimit = 1;
    } else {
        retryLimit = configuration.RETRY_LIMIT;
    }
    this.backoffInterval = configuration.BACKOFF_INTERVAL;
}

@Override
public IRecordProcessor createProcessor() {
    try {
        IBuffer<T> buffer = pipeline.getBuffer(configuration);
        IEmitter<U> emitter = pipeline.getEmitter(configuration);
        ITransformerBase<T, U> transformer = pipeline.getTransformer(configuration);
        IFilter<T> filter = pipeline.getFilter(configuration);
        KinesisConnectorRecordProcessor<T, U> processor =
                new KinesisConnectorRecordProcessor<T, U>(buffer, filter, emitter, transformer, configuration);
        return processor;
    } catch (Throwable t) {
        throw new RuntimeException(t);
    }
}

@Override
public IEmitter<String> getEmitter(KinesisConnectorConfiguration configuration) {
    return new SumologicEmitter(configuration);
}

@Override
public IEmitter<byte[]> getEmitter(KinesisConnectorConfiguration configuration) {
    return new S3Emitter(configuration);
}

@Override
public IEmitter<Map<String,String>> getEmitter(KinesisConnectorConfiguration configuration) {
    return new HBaseEmitter((EMRHBaseKinesisConnectorConfiguration) configuration);
}