test_class.declarator: LOCALHOST = "127.0.0.1"
test_class.modifier: private static final
test_class.original_string: private static final String LOCALHOST = "127.0.0.1";
test_class.type: String
test_class.var_name: LOCALHOST
test_class.file: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/source/SocketTextStreamFunctionTest.java
test_class.identifier: SocketTextStreamFunctionTest
test_case.body: @Test
  	public void testSocketSourceSimpleOutput() throws Exception {
  		ServerSocket server = new ServerSocket(0);
  		Socket channel = null;
  		
  		try {
  			SocketTextStreamFunction source = new SocketTextStreamFunction(LOCALHOST, server.getLocalPort(), '\n', 0);
  	
  			SocketSourceThread runner = new SocketSourceThread(source, "test1", "check");
  			runner.start();
  	
  			channel = server.accept();
  			OutputStreamWriter writer = new OutputStreamWriter(channel.getOutputStream());
  			
  			writer.write("test1\n");
  			writer.write("check\n");
  			writer.flush();
  			runner.waitForNumElements(2);
  
  			runner.cancel();
  			runner.interrupt();
  			
  			runner.waitUntilDone();
  			
  			channel.close();
  		}
  		finally {
  			if (channel != null) {
  				IOUtils.closeQuietly(channel);
  			}
  			IOUtils.closeQuietly(server);
  		}
  	}
test_case.class_method_signature: SocketTextStreamFunctionTest.testSocketSourceSimpleOutput()
test_case.full_signature: @Test public void testSocketSourceSimpleOutput()
test_case.identifier: testSocketSourceSimpleOutput
test_case.invocations: getLocalPort start accept getOutputStream write write flush waitForNumElements cancel interrupt waitUntilDone close closeQuietly closeQuietly
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testSocketSourceSimpleOutput()
focal_class.declarator: serialVersionUID = 1L
focal_class.modifier: private static final
focal_class.original_string: private static final long serialVersionUID = 1L;
focal_class.type: long
focal_class.var_name: serialVersionUID
focal_class.declarator: LOG = LoggerFactory.getLogger(SocketTextStreamFunction.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger LOG = LoggerFactory.getLogger(SocketTextStreamFunction.class);
focal_class.type: Logger
focal_class.var_name: LOG
focal_class.declarator: DEFAULT_CONNECTION_RETRY_SLEEP = 500
focal_class.modifier: private static final
focal_class.original_string: private static final int DEFAULT_CONNECTION_RETRY_SLEEP = 500;
focal_class.type: int
focal_class.var_name: DEFAULT_CONNECTION_RETRY_SLEEP
focal_class.declarator: CONNECTION_TIMEOUT_TIME = 0
focal_class.modifier: private static final
focal_class.original_string: private static final int CONNECTION_TIMEOUT_TIME = 0;
focal_class.type: int
focal_class.var_name: CONNECTION_TIMEOUT_TIME
focal_class.declarator: hostname
focal_class.modifier: private final
focal_class.original_string: private final String hostname;
focal_class.type: String
focal_class.var_name: hostname
focal_class.declarator: port
focal_class.modifier: private final
focal_class.original_string: private final int port;
focal_class.type: int
focal_class.var_name: port
focal_class.declarator: delimiter
focal_class.modifier: private final
focal_class.original_string: private final char delimiter;
focal_class.type: char
focal_class.var_name: delimiter
focal_class.declarator: maxNumRetries
focal_class.modifier: private final
focal_class.original_string: private final long maxNumRetries;
focal_class.type: long
focal_class.var_name: maxNumRetries
focal_class.declarator: delayBetweenRetries
focal_class.modifier: private final
focal_class.original_string: private final long delayBetweenRetries;
focal_class.type: long
focal_class.var_name: delayBetweenRetries
focal_class.declarator: currentSocket
focal_class.modifier: private transient
focal_class.original_string: private transient Socket currentSocket;
focal_class.type: Socket
focal_class.var_name: currentSocket
focal_class.declarator: isRunning = true
focal_class.modifier: private volatile
focal_class.original_string: private volatile boolean isRunning = true;
focal_class.type: boolean
focal_class.var_name: isRunning
focal_class.file: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/SocketTextStreamFunction.java
focal_class.identifier: SocketTextStreamFunction
focal_class.interfaces: implements SourceFunction<String>
focal_class.class_method_signature: SocketTextStreamFunction.SocketTextStreamFunction(String hostname, int port, char delimiter, long maxNumRetries)
focal_class.full_signature: public  SocketTextStreamFunction(String hostname, int port, char delimiter, long maxNumRetries)
focal_class.identifier: SocketTextStreamFunction
focal_class.modifiers: public
focal_class.parameters: (String hostname, int port, char delimiter, long maxNumRetries)
focal_class.signature:  SocketTextStreamFunction(String hostname, int port, char delimiter, long maxNumRetries)
focal_class.class_method_signature: SocketTextStreamFunction.SocketTextStreamFunction(String hostname, int port, char delimiter, long maxNumRetries, long delayBetweenRetries)
focal_class.full_signature: public  SocketTextStreamFunction(String hostname, int port, char delimiter, long maxNumRetries, long delayBetweenRetries)
focal_class.identifier: SocketTextStreamFunction
focal_class.modifiers: public
focal_class.parameters: (String hostname, int port, char delimiter, long maxNumRetries, long delayBetweenRetries)
focal_class.signature:  SocketTextStreamFunction(String hostname, int port, char delimiter, long maxNumRetries, long delayBetweenRetries)
focal_class.class_method_signature: SocketTextStreamFunction.run(SourceContext<String> ctx)
focal_class.full_signature: @Override public void run(SourceContext<String> ctx)
focal_class.identifier: run
focal_class.modifiers: @Override public
focal_class.parameters: (SourceContext<String> ctx)
focal_class.return: void
focal_class.signature: void run(SourceContext<String> ctx)
focal_class.class_method_signature: SocketTextStreamFunction.cancel()
focal_class.full_signature: @Override public void cancel()
focal_class.identifier: cancel
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void cancel()
focal_method.body: @Override
  	public void cancel() {
  		isRunning = false;
  		
  		// we need to close the socket as well, because the Thread.interrupt() function will
  		// not wake the thread in the socketStream.read() method when blocked.
  		Socket theSocket = this.currentSocket;
  		if (theSocket != null) {
  			IOUtils.closeSocket(theSocket);
  		}
  	}
focal_method.class_method_signature: SocketTextStreamFunction.cancel()
focal_method.full_signature: @Override public void cancel()
focal_method.identifier: cancel
focal_method.invocations: closeSocket
focal_method.modifiers: @Override public
focal_method.parameters: ()
focal_method.return: void
focal_method.signature: void cancel()