package com.customsource;

import java.io.IOException;
import java.net.ServerSocket;
import java.util.List;

import org.apache.flink.api.connector.source.SplitEnumerator;
import org.apache.flink.api.connector.source.SplitEnumeratorContext;

public class SocketEnumerator implements SplitEnumerator<SocketAsSplit, SocketAsEnum> {
	
	private final int port;
    private transient ServerSocket serverSocket;
    private final SplitEnumeratorContext<SocketAsSplit> context;
    
    public SocketEnumerator(int port, SplitEnumeratorContext<SocketAsSplit> context) {
		super();
		this.port = port;
		//this.serverSocket = serverSocket;
		this.context = context;
	}

	@Override
	public void start() {
		// TODO Auto-generated method stub
		try {
			this.serverSocket = new ServerSocket(this.port);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}

	@Override
	public void handleSplitRequest(int paramInt, String paramString) {
		// TODO Auto-generated method stub
		try {
			this.context.assignSplit(new SocketAsSplit(this.serverSocket.accept()), paramInt);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}

	@Override
	public void addSplitsBack(List<SocketAsSplit> paramList, int paramInt) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void addReader(int paramInt) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public SocketAsEnum snapshotState(long paramLong) throws Exception {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public void close() throws IOException {
		// TODO Auto-generated method stub
		
	}

}