@Override
	public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo, StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
		ServersocketSourceTableInfo serversocketSourceTableInfo = (ServersocketSourceTableInfo) sourceTableInfo;
		String tableName = serversocketSourceTableInfo.getName();

		TypeInformation[] types = new TypeInformation[serversocketSourceTableInfo.getFields().length];
		for (int i = 0; i < serversocketSourceTableInfo.getFieldClasses().length; i++) {
			types[i] = TypeInformation.of(serversocketSourceTableInfo.getFieldClasses()[i]);
		}
		TypeInformation typeInformation = new RowTypeInfo(types, serversocketSourceTableInfo.getFields());
		String fields = StringUtils.join(serversocketSourceTableInfo.getFields(), ",");

		CustomerSocketTextStreamFunction customerSocketTextStreamFunction = new CustomerSocketTextStreamFunction(serversocketSourceTableInfo, typeInformation,
				serversocketSourceTableInfo.getPhysicalFields(), serversocketSourceTableInfo.getFieldExtraInfoList());

		DataStreamSource serversocketSource = env.addSource(customerSocketTextStreamFunction, tableName, typeInformation);

		return tableEnv.fromDataStream(serversocketSource, fields);
	}
Reason:
Date: 2018/12/18
Company: www.dtstack.com

@author maqi
serversocket/serversocket-source/src/main/java/com/dtstack/flink/sql/source/serversocket/ServersocketSource.java