package com.informatica.vds.plugin.custom;

import java.io.IOException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import com.informatica.vds.api.VDSConfiguration;
import com.informatica.vds.api.VDSEventList;
import com.informatica.vds.api.VDSSource;

public class SimpleSource implements VDSSource {
	
	private String message;
	private BlockingQueue queue;
	private long count;
	
	@Override
	public void close() throws IOException {
		System.out.println("Closing custom source...");
	}

	@Override
	public void open(VDSConfiguration arg0) throws Exception {
		System.out.println("Opening custom source...");
		message = arg0.getString("message");
        
		queue = new LinkedBlockingQueue(100);
		
		for (int i = 0; i < 100; i++) {
			queue.offer(message + i);
		}		
	}

	@Override
	public void read(VDSEventList arg0) throws Exception {
		System.out.println("Reading source...");
		Thread.sleep(100);
		String message = (String) queue.poll();
		if (message == null) {
			return;
		}		
		arg0.addEvent(message.getBytes(), message.getBytes().length);				
	}
}