flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-libraries/flink-python/src/main/java/org/apache/flink/python/api/streaming/data/PythonStreamer.java
/**
	 * Sends all broadcast-variables encoded in the configuration to the external process.
	 *
	 * @param config configuration object containing broadcast-variable count and names
	 * @throws IOException
	 */
Sends all broadcast-variables encoded in the configuration to the external process.
@param config configuration object containing broadcast-variable count and names
@throws IOException
Sends all broadcast - variables encoded in the configuration to the external process . @param config configuration object containing broadcast - variable count and names @throws IOException
public final void sendBroadCastVariables(Configuration config) throws IOException {
		try {
			int broadcastCount = config.getInteger(PLANBINDER_CONFIG_BCVAR_COUNT, 0);
			String[] names = new String[broadcastCount];
			for (int x = 0; x < names.length; x++) {
				names[x] = config.getString(PLANBINDER_CONFIG_BCVAR_NAME_PREFIX + x, null);
			}
			out.write(new IntSerializer().serializeWithoutTypeInfo(broadcastCount));
			StringSerializer stringSerializer = new StringSerializer();
			for (String name : names) {
				Iterator<byte[]> bcv = function.getRuntimeContext().<byte[]>getBroadcastVariable(name).iterator();
				out.write(stringSerializer.serializeWithoutTypeInfo(name));
				while (bcv.hasNext()) {
					out.writeByte(1);
					out.write(bcv.next());
				}
				out.writeByte(0);
			}
		} catch (SocketTimeoutException ignored) {
			throw new RuntimeException("External process for task " + function.getRuntimeContext().getTaskName() + " stopped responding." + msg);
		}
	}
public final void sendBroadCastVariables ( Configuration config ) throws IOException { try { int broadcastCount = config . getInteger ( PLANBINDER_CONFIG_BCVAR_COUNT , 0 ) ; String [ ] names = new String [ broadcastCount ] ; for ( int x = 0 ; x < names . length ; x ++ ) { names [ x ] = config . getString ( PLANBINDER_CONFIG_BCVAR_NAME_PREFIX + x , null ) ; } out . write ( new IntSerializer ( ) . serializeWithoutTypeInfo ( broadcastCount ) ) ; StringSerializer stringSerializer = new StringSerializer ( ) ; for ( String name : names ) { Iterator < byte [ ] > bcv = function . getRuntimeContext ( ) . < byte [ ] > getBroadcastVariable ( name ) . iterator ( ) ; out . write ( stringSerializer . serializeWithoutTypeInfo ( name ) ) ; while ( bcv . hasNext ( ) ) { out . writeByte ( 1 ) ; out . write ( bcv . next ( ) ) ; } out . writeByte ( 0 ) ; } } catch ( SocketTimeoutException ignored ) { throw new RuntimeException ( "External process for task " + function . getRuntimeContext ( ) . getTaskName ( ) + " stopped responding." + msg ) ; } }
Sends all broadcast-variables encoded in the configuration to the external process.
Sends all broadcast - variables encoded in the configuration to the external process .