[GitHub] WillemJiang commented on a change in pull request #750: [SCB-484] Servlet rest support download

[GitBox <gi...@apache.org>]

WillemJiang commented on a change in pull request #750: [SCB-484] Servlet rest support download
URL: https://github.com/apache/incubator-servicecomb-java-chassis/pull/750#discussion_r193696911
 
 

 ##########
 File path: foundations/foundation-vertx/src/main/java/org/apache/servicecomb/foundation/vertx/stream/OutputStreamToWriteStream.java
 ##########
 @@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.servicecomb.foundation.vertx.stream;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.apache.servicecomb.foundation.common.io.AsyncCloseable;
+
+import io.vertx.core.Context;
+import io.vertx.core.Future;
+import io.vertx.core.Handler;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.streams.WriteStream;
+
+/**
+ * for pump from a readStream
+ */
+public class OutputStreamToWriteStream implements WriteStream<Buffer>, AsyncCloseable<Void> {
+  private static final int DEFAULT_MAX_BUFFERS = 4;
+
+  private static final int SMALLEST_MAX_BUFFERS = 2;
+
+  private OutputStream outputStream;
+
+  private Context context;
+
+  private boolean autoCloseOutputStream;
+
+  private Handler<Throwable> exceptionHandler;
+
+  // resume readStream
+  private Handler<Void> drainHandler;
+
+  // when invoke close, but outputStream not write all data, must put close logic to closedDeferred
+  private Runnable closedDeferred;
+
+  private boolean closed;
+
+  // buffers.size() need to loop all node, and maybe result is not correct in concurrent condition
+  // we just need to flow control by pump, so use another size
+  private Queue<Buffer> buffers = new ConcurrentLinkedQueue<>();
+
+  private int currentBufferCount;
+
+  // just indicate if buffers is full, not control add logic
+  // must >= SMALLEST_MAX_BUFFERS
+  // if < SMALLEST_MAX_BUFFERS, then maxBuffers will be SMALLEST_MAX_BUFFERS
+  private int maxBuffers = DEFAULT_MAX_BUFFERS;
+
+  // if currentBufferCount <= drainMark, will invoke drainHandler to resume readStream
+  private int drainMark = maxBuffers / 2;
+
+  public OutputStreamToWriteStream(Context context, OutputStream outputStream,
+      boolean autoCloseOutputStream) {
+    this.context = context;
+    this.outputStream = outputStream;
+    this.autoCloseOutputStream = autoCloseOutputStream;
+  }
+
+  @Override
+  public WriteStream<Buffer> exceptionHandler(Handler<Throwable> handler) {
+    this.exceptionHandler = handler;
+    return this;
+  }
+
+  private void handleException(Throwable t) {
+    if (exceptionHandler != null) {
+      exceptionHandler.handle(t);
+    }
+  }
+
+  @Override
+  public synchronized WriteStream<Buffer> write(Buffer data) {
+    currentBufferCount++;
+    buffers.add(data);
+    context.executeBlocking(this::writeInWorker,
+        true,
+        ar -> {
+          if (ar.failed()) {
+            handleException(ar.cause());
+          }
+        });
+    return this;
+  }
+
+  protected void writeInWorker(Future<Object> future) {
+    while (true) {
+      Buffer buffer = buffers.poll();
+      if (buffer == null) {
+        future.complete();
+        return;
+      }
+
+      try {
+        outputStream.write(buffer.getBytes());
+
+        synchronized (OutputStreamToWriteStream.this) {
+          currentBufferCount--;
+          Runnable action = (currentBufferCount == 0 && closedDeferred != null) ? closedDeferred : this::checkDrained;
+          action.run();
+        }
+      } catch (IOException e) {
+        currentBufferCount--;
 
 Review comment:
   Here is no sychronized.   If we just need to make sure currentBufferCount is automatic, we don't need to use the synchroized to do this job. 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services