| /* |
| * Copyright 2014, Google Inc. All rights reserved. |
| * |
| * Redistribution and use in source and binary forms, with or without |
| * modification, are permitted provided that the following conditions are |
| * met: |
| * |
| * * Redistributions of source code must retain the above copyright |
| * notice, this list of conditions and the following disclaimer. |
| * * Redistributions in binary form must reproduce the above |
| * copyright notice, this list of conditions and the following disclaimer |
| * in the documentation and/or other materials provided with the |
| * distribution. |
| * |
| * * Neither the name of Google Inc. nor the names of its |
| * contributors may be used to endorse or promote products derived from |
| * this software without specific prior written permission. |
| * |
| * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
| * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
| * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
| * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
| * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
| * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
| * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
| * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
| * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
| * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
| * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
| */ |
| |
| package io.grpc.internal; |
| |
| import static com.google.common.base.Preconditions.checkArgument; |
| import static com.google.common.base.Preconditions.checkNotNull; |
| import static com.google.common.base.Preconditions.checkState; |
| |
| import com.google.common.base.MoreObjects; |
| import com.google.common.base.Preconditions; |
| import io.grpc.Metadata; |
| import io.grpc.Status; |
| import java.io.InputStream; |
| import java.util.logging.Level; |
| import java.util.logging.Logger; |
| |
| /** |
| * The abstract base class for {@link ClientStream} implementations. |
| */ |
| public abstract class AbstractClientStream extends AbstractStream |
| implements ClientStream { |
| |
| private static final Logger log = Logger.getLogger(AbstractClientStream.class.getName()); |
| |
| private ClientStreamListener listener; |
| private boolean listenerClosed; |
| |
| // Stored status & trailers to report when deframer completes or |
| // transportReportStatus is directly called. |
| private Status status; |
| private Metadata trailers; |
| private Runnable closeListenerTask; |
| private volatile boolean cancelled; |
| |
| protected AbstractClientStream(WritableBufferAllocator bufferAllocator, int maxMessageSize, |
| StatsTraceContext statsTraceCtx) { |
| super(bufferAllocator, maxMessageSize, statsTraceCtx); |
| } |
| |
| @Override |
| public void setMaxInboundMessageSize(int maxSize) { |
| setMaxInboundMessageSizeProtected(maxSize); |
| } |
| |
| @Override |
| public void setMaxOutboundMessageSize(int maxSize) { |
| setMaxOutboundMessageSizeProtected(maxSize); |
| } |
| |
| @Override |
| protected final ClientStreamListener listener() { |
| return listener; |
| } |
| |
| /** |
| * Indicates that the stream may proceed to do traffic. Sub classes should override and invoke |
| * this method after they have been created. |
| */ |
| @Override |
| public void start(ClientStreamListener listener) { |
| checkState(this.listener == null, "stream already started"); |
| this.listener = checkNotNull(listener, "listener"); |
| } |
| |
| @Override |
| protected void receiveMessage(InputStream is) { |
| if (!listenerClosed) { |
| checkState(listener != null, "stream not started"); |
| listener.messageRead(is); |
| } |
| } |
| |
| /** |
| * The transport implementation has detected a protocol error on the stream. Transports are |
| * responsible for properly closing streams when protocol errors occur. |
| * |
| * @param errorStatus the error to report |
| * @param metadata any metadata received |
| */ |
| protected void inboundTransportError(Status errorStatus, Metadata metadata) { |
| Preconditions.checkNotNull(metadata, "metadata"); |
| if (inboundPhase() == Phase.STATUS) { |
| log.log(Level.INFO, "Received transport error on closed stream {0} {1}", |
| new Object[]{id(), errorStatus}); |
| return; |
| } |
| // For transport errors we immediately report status to the application layer |
| // and do not wait for additional payloads. |
| transportReportStatus(errorStatus, false, metadata); |
| } |
| |
| /** |
| * Called by transport implementations when they receive headers. |
| * |
| * @param headers the parsed headers |
| */ |
| protected void inboundHeadersReceived(Metadata headers) { |
| checkState(listener != null, "stream not started"); |
| if (inboundPhase() == Phase.STATUS) { |
| log.log(Level.INFO, "Received headers on closed stream {0} {1}", |
| new Object[]{id(), headers}); |
| } |
| |
| inboundPhase(Phase.MESSAGE); |
| listener.headersRead(headers); |
| } |
| |
| /** |
| * Processes the contents of a received data frame from the server. |
| * |
| * @param frame the received data frame. Its ownership is transferred to this method. |
| */ |
| protected void inboundDataReceived(ReadableBuffer frame) { |
| Preconditions.checkNotNull(frame, "frame"); |
| boolean needToCloseFrame = true; |
| try { |
| if (inboundPhase() == Phase.STATUS) { |
| return; |
| } |
| if (inboundPhase() == Phase.HEADERS) { |
| // Have not received headers yet so error |
| inboundTransportError(Status.INTERNAL |
| .withDescription("headers not received before payload"), new Metadata()); |
| return; |
| } |
| inboundPhase(Phase.MESSAGE); |
| |
| needToCloseFrame = false; |
| deframe(frame, false); |
| } finally { |
| if (needToCloseFrame) { |
| frame.close(); |
| } |
| } |
| } |
| |
| @Override |
| protected void inboundDeliveryPaused() { |
| runCloseListenerTask(); |
| } |
| |
| @Override |
| protected final void deframeFailed(Throwable cause) { |
| cancel(Status.INTERNAL.withDescription("Exception deframing message").withCause(cause)); |
| } |
| |
| /** |
| * Processes the trailers and status from the server. |
| * |
| * @param trailers the received trailers |
| * @param status the status extracted from the trailers |
| */ |
| protected void inboundTrailersReceived(Metadata trailers, Status status) { |
| Preconditions.checkNotNull(trailers, "trailers"); |
| if (inboundPhase() == Phase.STATUS) { |
| log.log(Level.INFO, "Received trailers on closed stream {0}\n {1}\n {2}", |
| new Object[]{id(), status, trailers}); |
| } |
| // Stash the status & trailers so they can be delivered by the deframer calls |
| // remoteEndClosed |
| this.status = status; |
| this.trailers = trailers; |
| deframe(ReadableBuffers.empty(), true); |
| } |
| |
| @Override |
| protected void remoteEndClosed() { |
| transportReportStatus(status, true, trailers); |
| } |
| |
| @Override |
| protected final void internalSendFrame(WritableBuffer frame, boolean endOfStream, boolean flush) { |
| Preconditions.checkArgument(frame != null || endOfStream, "null frame before EOS"); |
| sendFrame(frame, endOfStream, flush); |
| } |
| |
| /** |
| * Sends an outbound frame to the remote end point. |
| * |
| * @param frame a buffer containing the chunk of data to be sent or {@code null} if the framer is |
| * closing and has no data to send. |
| * @param endOfStream if {@code true} indicates that no more data will be sent on the stream by |
| * this endpoint. |
| * @param flush {@code true} if more data may not be arriving soon |
| */ |
| protected abstract void sendFrame(WritableBuffer frame, boolean endOfStream, boolean flush); |
| |
| /** |
| * Report stream closure with status to the application layer if not already reported. This method |
| * must be called from the transport thread. |
| * |
| * @param newStatus the new status to set |
| * @param stopDelivery if {@code true}, interrupts any further delivery of inbound messages that |
| * may already be queued up in the deframer. If {@code false}, the listener will be |
| * notified immediately after all currently completed messages in the deframer have been |
| * delivered to the application. |
| * @param trailers new instance of {@code Trailers}, either empty or those returned by the server |
| */ |
| public void transportReportStatus(final Status newStatus, boolean stopDelivery, |
| final Metadata trailers) { |
| Preconditions.checkNotNull(newStatus, "newStatus"); |
| |
| boolean closingLater = closeListenerTask != null && !stopDelivery; |
| if (listenerClosed || closingLater) { |
| // We already closed (or are about to close) the listener. |
| return; |
| } |
| |
| inboundPhase(Phase.STATUS); |
| status = newStatus; |
| closeListenerTask = null; |
| |
| // Determine if the deframer is stalled (i.e. currently has no complete messages to deliver). |
| boolean deliveryStalled = isDeframerStalled(); |
| |
| if (stopDelivery || deliveryStalled) { |
| // Close the listener immediately. |
| closeListener(newStatus, trailers); |
| } else { |
| // Delay close until inboundDeliveryStalled() |
| closeListenerTask = newCloseListenerTask(newStatus, trailers); |
| } |
| } |
| |
| /** |
| * Creates a new {@link Runnable} to close the listener with the given status/trailers. |
| */ |
| private Runnable newCloseListenerTask(final Status status, final Metadata trailers) { |
| return new Runnable() { |
| @Override |
| public void run() { |
| closeListener(status, trailers); |
| } |
| }; |
| } |
| |
| /** |
| * Closes the listener if not previously closed. |
| * |
| * @throws IllegalStateException if the call has not yet been started. |
| */ |
| private void closeListener(Status newStatus, Metadata trailers) { |
| checkState(listener != null, "stream not started"); |
| if (!listenerClosed) { |
| listenerClosed = true; |
| closeDeframer(); |
| listener.closed(newStatus, trailers); |
| } |
| } |
| |
| /** |
| * Executes the pending listener close task, if one exists. |
| */ |
| private void runCloseListenerTask() { |
| if (closeListenerTask != null) { |
| closeListenerTask.run(); |
| closeListenerTask = null; |
| } |
| } |
| |
| @Override |
| public final void halfClose() { |
| if (outboundPhase(Phase.STATUS) != Phase.STATUS) { |
| closeFramer(); |
| } |
| } |
| |
| /** |
| * Cancel the stream. Called by the application layer, never called by the transport. |
| */ |
| @Override |
| public final void cancel(Status reason) { |
| checkArgument(!reason.isOk(), "Should not cancel with OK status"); |
| cancelled = true; |
| sendCancel(reason); |
| dispose(); |
| } |
| |
| @Override |
| public final boolean isReady() { |
| return !cancelled && super.isReady(); |
| } |
| |
| /** |
| * Cancel the stream and send a stream cancellation message to the remote server, if necessary. |
| * Can be called by either the application or transport layers. This method is safe to be called |
| * at any time and multiple times. |
| */ |
| protected abstract void sendCancel(Status reason); |
| |
| @Override |
| protected MoreObjects.ToStringHelper toStringHelper() { |
| MoreObjects.ToStringHelper toStringHelper = super.toStringHelper(); |
| if (status != null) { |
| toStringHelper.add("status", status); |
| } |
| return toStringHelper; |
| } |
| |
| @Override |
| public boolean isClosed() { |
| return super.isClosed() || listenerClosed; |
| } |
| } |