package tunan.hadoop.flume.sources;

import org.apache.flume.Context;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.source.AbstractSource;

public class MySource extends AbstractSource implements Configurable, PollableSource {

    // 自定义属性，前缀和后缀
    private String prefix;
    private String suffix;

    // 处理Event
    @Override
    public Status process() throws EventDeliveryException {
        // 自定义状态属性
        Status status = null;

        // 模拟产生数据
        for (int i = 0; i < 100; i++) {
            // 创建Event
            SimpleEvent event = new SimpleEvent();

            // 把数据设置到Body中去，注意header为空
            event.setBody((prefix + i + suffix).getBytes());
            // 开始处理Evetn
            getChannelProcessor().processEvent(event);
        }

        try {
            // 成功
            status = Status.READY;
        } catch (Exception e) {
            e.printStackTrace();
            // 失败，回退
            status = Status.BACKOFF;
        }

        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 返回状态的结果
        return status;
    }


    // 获取Agent中传入的参数信息
    @Override
    public void configure(Context context) {
        this.prefix = context.getString("prefix","TUNAN");
        this.suffix = context.getString("suffix");
    }
}

package tunan.hadoop.flume.sink;

import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MySink extends AbstractSink implements Configurable {

    // 拿到logger
    private static final Logger logger =  LoggerFactory.getLogger(MySink.class);

    // 属性作为参数的前缀和后缀
    private String prefix;
    private String suffix;

    // 从channel中获取数据发送到目的地
    @Override
    public Status process() throws EventDeliveryException {
        Status status;

        // 获取channel
        Channel ch = getChannel();
        // 获取事物
        Transaction txn = ch.getTransaction();
        // 开启事物
        txn.begin();
        try {
            // 拿到event
            Event event;

            do {   // 一直等待，直到拿到不为空的Event
                event = ch.take();
            } while ((event == null));

            // body是个字节数组转换成字符串
            String body = new String(event.getBody());

            // 控制台打印
            logger.error(prefix + body + suffix);

            // 提交事务
            txn.commit();
            status = Status.READY;
        } catch (Throwable t) {
            // 回滚事务
            txn.rollback();
            status = Status.BACKOFF;

            if (t instanceof Error) {
                throw (Error)t;
            }
        } finally {
            txn.close();
        }
        return status;
    }

    // 从Agent中拿到参数
    @Override
    public void configure(Context context) {
        this.prefix = context.getString("prefix","Tunan");
        this.suffix = context.getString("suffix");
    }
}

package tunan.hadoop.flume.interceptor;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

public class MyInterceptor implements Interceptor {
	// 自定义List用来处理批量Event
    private  List<Event> newEvents;

    // 初始化设置
    @Override
    public void initialize()
    {
        newEvents = new ArrayList<>();
    }

    // 拦截单个Event
    @Override
    public Event intercept(Event event) {
        Map<String, String> headers = event.getHeaders();
        String body = new String(event.getBody());
        if (body.contains("gifshow")) {
            // 自定义头信息
            headers.put("type", "gifshow");
        	// 自定义头信息
        } else {
            headers.put("type", "other");
        }

        return event;
    }

    // 拦截多个Event处理
    @Override
    public List<Event> intercept(List<Event> events) {
        // 每次进来初始化List
        newEvents.clear();
        Iterator<Event> iter = events.iterator();
        while (iter.hasNext()){
            Event next = iter.next();
            // event传递给单个处理，并添加到新的List
            newEvents.add(intercept(next));
        }

        // 返回拦截后的Event
        return newEvents;
    }

    @Override
    public void close() {

    }
    // 源码在 HostInterceptor，需要添加一个静态内部类，并且名为Builder
    public static class Builder implements Interceptor.Builder{

        @Override
        public Interceptor build() {
            return new MyInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }
}