}
Generated the stubs on both python and java sides.
import os
from concurrent import futures
import grpc
import time
import file_upload_pb2, file_upload_pb2_grpc
CHUNK_SIZE = 1024 * 1024 # 1MB
def get_file_chunks(filename):
with open(filename, 'rb') as f:
while True:
piece = f.read(CHUNK_SIZE);
if len(piece) == 0:
return
yield chunk_pb2.Chunk(buffer=piece)
def save_chunks_to_file(chunks, filename):
with open(filename, 'wb') as f:
for chunk in chunks:
f.write(chunk.buffer)
class FileServer(file_upload_pb2_grpc.FileServerServicer):
def __init__(self):
print("in Filervser")
class Servicer(file_upload_pb2_grpc.FileServerServicer):
def __init__(self):
print("in service")
self.tmp_file_name = 'stupid.poto'
def upload(self, request_iterator, context):
print("in upload")
save_chunks_to_file(request_iterator, self.tmp_file_name)
return file_upload_pb2.Reply(length=os.path.getsize(self.tmp_file_name))
print("in server")
self.server = grpc.server(futures.ThreadPoolExecutor(max_workers=1))
file_upload_pb2_grpc.add_FileServerServicer_to_server(Servicer(), self.server)
def start(self, port):
print("starting")
self.server.add_insecure_port(f'[::]:{port}')
self.server.start()
try:
while True:
time.sleep(60*60*24)
except KeyboardInterrupt:
self.server.stop(0)
if __name__ == '__main__':
FileServer().start(8888)
Java file
package test.fileupload;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.util.logging.Logger;
import com.google.protobuf.ByteString;
import test.fileupload.FileUpload.Reply;
import test.fileupload.FileUpload.Chunk;
public class FileUploadClient {
public static void main(String[] args) {
ManagedChannel mChannel= ManagedChannelBuilder.forAddress("localhost", 8888)
.usePlaintext().build();
FileServerGrpc.FileServerStub fileServerStub = FileServerGrpc.newStub(mChannel);
StreamObserver<Reply> streamReply = new StreamObserver<Reply>() {
@Override
public void onNext(Reply value) {
Logger.getAnonymousLogger().info("on Next");
}
@Override
public void onError(Throwable t) {
Logger.getAnonymousLogger().info("on Error");
}
@Override
public void onCompleted() {
Logger.getAnonymousLogger().info("on Completed");
}
};
StreamObserver<Chunk> chunkStream =fileServerStub.upload(streamReply);
try {
File file = new File("resizedCat.png");
if (file.exists() == false) {
Logger.getAnonymousLogger().info("File does not exist");
return;
}
BufferedInputStream bInputStream = new BufferedInputStream(new FileInputStream(file));
int bufferSize = 1024 * 1024; // 512k
byte[] buffer = new byte[bufferSize];
int tmp = 0;
int size = 0;
while ((tmp = bInputStream.read(buffer)) > 0) {
size += tmp;
ByteString byteString = ByteString.copyFrom(buffer);
Chunk req = Chunk.newBuilder().setBuffer(byteString).build();
chunkStream.onNext(req);
}
}catch (Exception e) {
Logger.getAnonymousLogger().info("Exception"+e.getMessage());
}
chunkStream.onCompleted();
}
}