core,netty: quick patch for setListener regression

resolves grpc/grpc#8715
now that setListener is called prior to
`JumpToApplicationThreadServerStreamListener` being completely ready to
use. We should not call `AbstractStream2#onStreamAllocated()` inside
`setListener()` anymore, but call it after `ServerImpl#streamCreated()`
is completed.

Author: dapengzhang0

core/src/main/java/io/grpc/internal/AbstractServerStream.java

@@ -84,7 +84,7 @@ protected interface Sink {
      * Tears down the stream, typically in the event of a timeout. This method may be called
      * multiple times and from any thread.
      *
-     * <p>This is a clone of {@link ServerStream#cancel()}.
+     * <p>This is a clone of {@link ServerStream#cancel(Status)}.
      */
     void cancel(Status status);
   }
@@ -189,11 +189,13 @@ protected TransportState(int maxMessageSize, StatsTraceContext statsTraceCtx) {
      * thread.
      */
     public final void setListener(ServerStreamListener listener) {
+      Preconditions.checkState(this.listener == null, "setListener should be called only once");
       this.listener = Preconditions.checkNotNull(listener, "listener");
+    }
 
-      // Now that the stream has actually been initialized, call the listener's onReady callback if
-      // appropriate.
-      onStreamAllocated();
+    @Override
+    public final void onStreamAllocated() {
+      super.onStreamAllocated();
     }
 
     @Override

core/src/main/java/io/grpc/internal/AbstractStream2.java

@@ -237,7 +237,7 @@ private boolean isReady() {
      * StreamListener#onReady()} handler if appropriate. This must be called from the transport
      * thread, since the listener may be called back directly.
      */
-    protected final void onStreamAllocated() {
+    protected void onStreamAllocated() {
       checkState(listener() != null);
       synchronized (onReadyLock) {
         checkState(!allocated, "Already allocated");

core/src/test/java/io/grpc/internal/AbstractServerStreamTest.java

@@ -107,17 +107,27 @@ public void completeWithoutClose() {
 
   @Test
   public void setListener_setOnlyOnce() {
-
     stream.transportState().setListener(new ServerStreamListenerBase());
     thrown.expect(IllegalStateException.class);
 
     stream.transportState().setListener(new ServerStreamListenerBase());
   }
 
   @Test
-  public void setListener_readyCalled() {
+  public void listenerReady_onlyOnce() {
+    stream.transportState().setListener(new ServerStreamListenerBase());
+    stream.transportState().onStreamAllocated();
+    thrown.expect(IllegalStateException.class);
+
+    stream.transportState().onStreamAllocated();
+  }
+
+
+  @Test
+  public void listenerReady_readyCalled() {
     ServerStreamListener streamListener = mock(ServerStreamListener.class);
     stream.transportState().setListener(streamListener);
+    stream.transportState().onStreamAllocated();
 
     verify(streamListener).onReady();
   }

netty/src/main/java/io/grpc/netty/NettyServerHandler.java

@@ -200,6 +200,7 @@ private void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers
       NettyServerStream stream = new NettyServerStream(ctx.channel(), state, attributes,
           statsTraceCtx);
       transportListener.streamCreated(stream, method, metadata);
+      state.onStreamAllocated();
       http2Stream.setProperty(streamKey, state);
 
     } catch (Http2Exception e) {

netty/src/test/java/io/grpc/netty/NettyServerStreamTest.java

@@ -298,6 +298,7 @@ public Object answer(InvocationOnMock invocation) throws Throwable {
     NettyServerStream stream = new NettyServerStream(channel, state, Attributes.EMPTY,
         statsTraceCtx);
     stream.transportState().setListener(serverListener);
+    state.onStreamAllocated();
     verify(serverListener, atLeastOnce()).onReady();
     verifyNoMoreInteractions(serverListener);
     return stream;