package com.dataartisans.flink.dataflow.translation.wrappers;

import org.apache.flink.core.memory.DataInputView;

import java.io.IOException;
import java.io.InputStream;

/**
 * Wrapper for {@link DataInputView}. We need this because Flink reads data using a
 * {@link org.apache.flink.core.memory.DataInputView} while
 * Dataflow {@link com.google.cloud.dataflow.sdk.coders.Coder}s expect an
 * {@link java.io.InputStream}.
 */
public class DataInputViewWrapper extends InputStream {

	private DataInputView inputView;

	public DataInputViewWrapper(DataInputView inputView) {
		this.inputView = inputView;
	}

	public void setInputView(DataInputView inputView) {
		this.inputView = inputView;
	}

	@Override
	public int read() throws IOException {
		return inputView.readByte();
	}

	@Override
	public int read(byte[] b, int off, int len) throws IOException {
		return inputView.read(b, off, len);
	}
}

@@ -108,16 +108,45 @@
 tView;%0A%0A
+import java.io.EOFException;%0A
 import j
@@ -764,32 +764,41 @@
 s IOException %7B%0A
+%09%09try %7B%0A%09
 %09%09return inputVi
@@ -808,15 +808,199 @@
 read
-Byte();
+UnsignedByte();%0A%09%09%7D catch (EOFException e) %7B%0A%09%09%09// translate between DataInput and InputStream,%0A%09%09%09// DataInput signals EOF by exception, InputStream does it by returning -1%0A%09%09%09return -1;%0A%09%09%7D
 %0A%09%7D%0A

Fix mismatch between DataInput and InputStream in DataInputViewWrapper