import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
private SiteToSiteClient getClient(String url, String port, boolean useCompression, SSLContext sslContext) {
  SiteToSiteClient.Builder configBuilder = new SiteToSiteClient.Builder()
      .url(url)
      .portName(port)
      .useCompression(useCompression);

  if (sslContext != null) {
      configBuilder = configBuilder.sslContext(sslContext);
  }

  final SiteToSiteClientConfig config = configBuilder.buildConfig();

  final SiteToSiteClient client = new SiteToSiteClient.Builder()
     .fromConfig(config)
     .build();

  return client;
}
 

import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

	SiteToSiteClientConfig clientConfig = new SiteToSiteClient.Builder()
			.url("http://localhost:8080/nifi")
			.portName("Data for Flink")
			.requestBatchCount(5)
			.buildConfig();

	SourceFunction<NiFiDataPacket> nifiSource = new NiFiSource(clientConfig);
	DataStream<NiFiDataPacket> streamSource = env.addSource(nifiSource).setParallelism(2);

	DataStream<String> dataStream = streamSource.map(new MapFunction<NiFiDataPacket, String>() {
		@Override
		public String map(NiFiDataPacket value) throws Exception {
			return new String(value.getContent(), Charset.defaultCharset());
		}
	});

	dataStream.print();
	env.execute();
}
 

import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

	SiteToSiteClientConfig clientConfig = new SiteToSiteClient.Builder()
			.url("http://localhost:8080/nifi")
			.portName("Data from Flink")
			.buildConfig();

	DataStreamSink<String> dataStream = env.fromElements("one", "two", "three", "four", "five", "q")
			.addSink(new NiFiSink<>(clientConfig, new NiFiDataPacketBuilder<String>() {
				@Override
				public NiFiDataPacket createNiFiDataPacket(String s, RuntimeContext ctx) {
					return new StandardNiFiDataPacket(s.getBytes(ConfigConstants.DEFAULT_CHARSET),
						new HashMap<String, String>());
				}
			}));

	env.execute();
}
 

import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

	SiteToSiteClientConfig clientConfig = new SiteToSiteClient.Builder()
			.url("http://localhost:8080/nifi")
			.portName("Data from Flink")
			.buildConfig();

	DataStreamSink<String> dataStream = env.fromElements("one", "two", "three", "four", "five", "q")
			.addSink(new NiFiSink<>(clientConfig, new NiFiDataPacketBuilder<String>() {
				@Override
				public NiFiDataPacket createNiFiDataPacket(String s, RuntimeContext ctx) {
					return new StandardNiFiDataPacket(s.getBytes(), new HashMap<String,String>());
				}
			}));

	env.execute();
}
 

import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
@Override
public void populateDAG(DAG dag, Configuration conf)
{
  final SiteToSiteClientConfig clientConfig = new SiteToSiteClient.Builder()
      .url("http://localhost:8080/nifi")
      .portName("Apex")
      .requestBatchCount(5)
      .buildConfig();

  final SiteToSiteClient.Builder builder = new SiteToSiteClient.Builder().fromConfig(clientConfig);

  final WindowDataManager windowDataManager = new WindowDataManager.NoopWindowDataManager();

  NiFiSinglePortInputOperator nifi = dag.addOperator("nifi", new NiFiSinglePortInputOperator(builder, windowDataManager));
  ConsoleOutputOperator console = dag.addOperator("console", new ConsoleOutputOperator());
  dag.addStream("nifi_console", nifi.outputPort, console.input).setLocality(null);
}
 

import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
@Override
public void populateDAG(DAG dag, Configuration conf)
{
  final SiteToSiteClientConfig clientConfig = new SiteToSiteClient.Builder()
      .url("http://localhost:8080/nifi")
      .portName("Apex")
      .buildConfig();

  final int batchSize = 1;
  final SiteToSiteClient.Builder builder = new SiteToSiteClient.Builder().fromConfig(clientConfig);
  final NiFiDataPacketBuilder<String> dataPacketBuilder = new StringNiFiDataPacketBuilder();
  final WindowDataManager windowDataManager = new WindowDataManager.NoopWindowDataManager();

  RandomEventGenerator rand = dag.addOperator("rand", new RandomEventGenerator());

  NiFiSinglePortOutputOperator nifi = dag.addOperator("nifi",
      new NiFiSinglePortOutputOperator(builder, dataPacketBuilder, windowDataManager, batchSize));

  dag.addStream("rand_nifi", rand.string_data, nifi.inputPort).setLocality(null);
}
 

import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
public NifiPublisher(final SiteToSiteClient client, List<ILoggingEvent> messagesList, int pushInterval,
		Fallback fallbackForError) {
	this.client = client;
	this.messagesList = messagesList;
	this.pushInterval = pushInterval;
	this.fallbackForError = fallbackForError;
}
 

import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
@Override
public void start() {
	//validate configuration
	if (!isConfigurationValid())
		return;

	client = new SiteToSiteClient.Builder().url(getUrl()).portIdentifier(getReceiverPortId())
			.portName(getReceiverPortName()).transportProtocol(getSiteToSiteTransportProtocol()).build();
	messagesList = new ArrayList<ILoggingEvent>();
	new Thread(new NifiPublisher(client, messagesList,getPushInterval(),new DefaultFallBackImpl())).start();
	super.start();
}
 

import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
private static SiteToSiteClientConfig getSourceConfig(WindowLogLevelCountProps props) {
    return new SiteToSiteClient.Builder()
            .url(props.getNifiUrl())
            .portName(props.getNifiInputPort())
            .requestBatchCount(props.getNifiRequestBatch())
            .buildConfig();
}
 

import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
private NiFiSinglePortInputOperator getNiFiInput(DAG dag, LogLevelProperties props, WindowDataManager windowDataManager) {
    final SiteToSiteClient.Builder inputConfig = new SiteToSiteClient.Builder()
            .url(props.getNifiUrl())
            .portName(props.getNifiInputPort())
            .requestBatchCount(props.getNifiRequestBatch());

    return dag.addOperator("nifi-in", new NiFiSinglePortInputOperator(inputConfig, windowDataManager));
}
 

import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
private NiFiSinglePortOutputOperator getNiFiOutput(DAG dag, LogLevelProperties props, WindowDataManager windowDataManager) {
    final SiteToSiteClient.Builder outputConfig = new SiteToSiteClient.Builder()
            .url(props.getNifiUrl())
            .portName(props.getNifiOutputPort());

    final int batchSize = 1;
    final NiFiDataPacketBuilder<LogLevels> dataPacketBuilder = new DictionaryBuilder(
            props.getWindowMillis(), props.getLogLevelThreshold());

    return dag.addOperator("nifi-out", new NiFiSinglePortOutputOperator(
            outputConfig, dataPacketBuilder, windowDataManager ,batchSize));
}
 

import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
private static SiteToSiteClientConfig getSourceConfig(LogLevelCountProperties props) {
    return new SiteToSiteClient.Builder()
            .url(props.getNifiUrl())
            .portName(props.getNifiInputPort())
            .requestBatchCount(props.getNifiRequestBatch())
            .buildConfig();
}
 

import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
/**
 * @param siteToSiteBuilder the builder for a NiFi SiteToSiteClient
 * @param dataPacketBuilder a builder to produce NiFiDataPackets from incoming data
 * @param windowDataManager  a WindowDataManager to save and load state for windows of tuples
 */
public AbstractNiFiOutputOperator(final SiteToSiteClient.Builder siteToSiteBuilder,
    final NiFiDataPacketBuilder<T> dataPacketBuilder, final WindowDataManager windowDataManager)
{
  this.siteToSiteBuilder = siteToSiteBuilder;
  this.dataPacketBuilder = dataPacketBuilder;
  this.windowDataManager = windowDataManager;
}
 

import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
/**
 * @param siteToSiteBuilder the builder for a NiFi SiteToSiteClient
 * @param windowDataManager a WindowDataManager to save and load state for windows of tuples
 */
public AbstractNiFiInputOperator(final SiteToSiteClient.Builder siteToSiteBuilder,
    final WindowDataManager windowDataManager)
{
  this.siteToSiteBuilder = siteToSiteBuilder;
  this.windowDataManager = windowDataManager;
}
 

import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
/**
 * @param siteToSiteBuilder the builder for a NiFi SiteToSiteClient
 * @param dataPacketBuilder a builder to produce NiFiDataPackets from incoming data
 * @param windowDataManager  a WindowDataManager to save and load state for windows of tuples
 * @param batchSize the maximum number of tuples to send to NiFi in a single transaction
 */
public NiFiSinglePortOutputOperator(
    final SiteToSiteClient.Builder siteToSiteBuilder,
    final NiFiDataPacketBuilder<T> dataPacketBuilder,
    final WindowDataManager windowDataManager,
    final int batchSize)
{
  super(siteToSiteBuilder, dataPacketBuilder, windowDataManager);
  this.inputPort = new BufferingInputPort(batchSize);
}
 

import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
private static SiteToSiteClientConfig getSinkConfig(WindowLogLevelCountProps props) {
    return new SiteToSiteClient.Builder()
            .url(props.getNifiUrl())
            .portName(props.getNifiOutputPort())
            .buildConfig();
}
 

import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
private static SiteToSiteClientConfig getSinkConfig(LogLevelCountProperties props) {
    return new SiteToSiteClient.Builder()
            .url(props.getNifiUrl())
            .portName(props.getNifiOutputPort())
            .buildConfig();
}
 

import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
@Override
public void open(Configuration parameters) throws Exception {
	super.open(parameters);
	client = new SiteToSiteClient.Builder().fromConfig(clientConfig).build();
}
 

import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
@Override
public void open(Configuration parameters) throws Exception {
	super.open(parameters);
	this.client = new SiteToSiteClient.Builder().fromConfig(clientConfig).build();
}
 

import org.apache.nifi.remote.client.SiteToSiteClient; //导入依赖的package包/类
/**
 *
 * @param siteToSiteBuilder the builder for a NiFi SiteToSiteClient
 * @param windowDataManager a WindowDataManager to save and load state for windows of tuples
 */
public NiFiSinglePortInputOperator(final SiteToSiteClient.Builder siteToSiteBuilder,
    final WindowDataManager windowDataManager)
{
  super(siteToSiteBuilder, windowDataManager);
}