#20469

Committed 10 Sep 2026 08:17AM UTC coverage: 89.337% (+0.06%) from 89.279%

Build # #20469

Build Type

push

github

Committed by

web-flow

Commit Message

api: Implement custom events framework in gRPC-Java server (#12980)

This adds triggerEvent/onEvent APIs to `ServerCall` and `ServerCall.Listener` routing them through `ServerStream` transport.

Coverage Stats

39067 of 43730 relevant lines covered (89.34%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

44.74

/../util/src/main/java/io/grpc/util/TransmitStatusRuntimeExceptionInterceptor.java

/*
 * Copyright 2017 The gRPC Authors
 *
 * Licensed 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 io.grpc.util;

import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import io.grpc.Attributes;
import io.grpc.ExperimentalApi;
import io.grpc.ForwardingServerCall;
import io.grpc.ForwardingServerCallListener;
import io.grpc.Metadata;
import io.grpc.ServerCall;
import io.grpc.ServerCallHandler;
import io.grpc.ServerInterceptor;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.internal.SerializingExecutor;
import java.util.concurrent.ExecutionException;
import javax.annotation.Nullable;

/**
 * A class that intercepts uncaught exceptions of type {@link StatusRuntimeException} and handles
 * them by closing the {@link ServerCall}, and transmitting the exception's status and metadata
 * to the client.
 *
 * <p>Without this interceptor, gRPC will strip all details and close the {@link ServerCall} with
 * a generic {@link Status#UNKNOWN} code.
 *
 * <p>Security warning: the {@link Status} and {@link Metadata} may contain sensitive server-side
 * state information, and generally should not be sent to clients. Only install this interceptor
 * if all clients are trusted.
 */
@ExperimentalApi("https://github.com/grpc/grpc-java/issues/2189")
public final class TransmitStatusRuntimeExceptionInterceptor implements ServerInterceptor {
  private TransmitStatusRuntimeExceptionInterceptor() {
  }

  public static ServerInterceptor instance() {
    return new TransmitStatusRuntimeExceptionInterceptor();
  }

  @Override
  public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
      ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
    final ServerCall<ReqT, RespT> serverCall = new SerializingServerCall<>(call);
    ServerCall.Listener<ReqT> listener = next.startCall(serverCall, headers);
    return new ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT>(listener) {
      @Override
      public void onMessage(ReqT message) {
        try {
          super.onMessage(message);
        } catch (StatusRuntimeException e) {
          closeWithException(e);
        }
      }

      @Override
      public void onHalfClose() {
        try {
          super.onHalfClose();
        } catch (StatusRuntimeException e) {
          closeWithException(e);
        }
      }

      @Override
      public void onCancel() {
        try {
          super.onCancel();
        } catch (StatusRuntimeException e) {
          closeWithException(e);
        }
      }

      @Override
      public void onComplete() {
        try {
          super.onComplete();
        } catch (StatusRuntimeException e) {
          closeWithException(e);
        }
      }

      @Override
      public void onReady() {
        try {
          super.onReady();
        } catch (StatusRuntimeException e) {
          closeWithException(e);
        }
      }

      @Override
      public void onEvent(Object event) {
        try {
          super.onEvent(event);
        } catch (StatusRuntimeException e) {
          closeWithException(e);
        }
      }

      private void closeWithException(StatusRuntimeException t) {
        Metadata metadata = t.getTrailers();
        if (metadata == null) {
          metadata = new Metadata();
        }
        serverCall.close(t.getStatus(), metadata);
      }
    };
  }

  /**
   * A {@link ServerCall} that wraps around a non thread safe delegate and provides thread safe
   * access by serializing everything on an executor.
   */
  private static class SerializingServerCall<ReqT, RespT> extends
      ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT> {
    private static final String ERROR_MSG = "Encountered error during serialized access";
    private final SerializingExecutor serializingExecutor =
        new SerializingExecutor(MoreExecutors.directExecutor());
    private boolean closeCalled = false;

    SerializingServerCall(ServerCall<ReqT, RespT> delegate) {
      super(delegate);
    }

    @Override
    public void sendMessage(final RespT message) {
      serializingExecutor.execute(new Runnable() {
        @Override
        public void run() {
          SerializingServerCall.super.sendMessage(message);
        }
      });
    }

    @Override
    public void request(final int numMessages) {
      serializingExecutor.execute(new Runnable() {
        @Override
        public void run() {
          SerializingServerCall.super.request(numMessages);
        }
      });
    }

    @Override
    public void sendHeaders(final Metadata headers) {
      serializingExecutor.execute(new Runnable() {
        @Override
        public void run() {
          SerializingServerCall.super.sendHeaders(headers);
        }
      });
    }

    @Override
    public void close(final Status status, final Metadata trailers) {
      serializingExecutor.execute(new Runnable() {
        @Override
        public void run() {
          if (!closeCalled) {
            closeCalled = true;

            SerializingServerCall.super.close(status, trailers);
          }
        }
      });
    }

    @Override
    public boolean isReady() {
      final SettableFuture<Boolean> retVal = SettableFuture.create();
      serializingExecutor.execute(new Runnable() {
        @Override
        public void run() {
          retVal.set(SerializingServerCall.super.isReady());
        }
      });
      try {
        return retVal.get();
      } catch (InterruptedException e) {
        throw new RuntimeException(ERROR_MSG, e);
      } catch (ExecutionException e) {
        throw new RuntimeException(ERROR_MSG, e);
      }
    }

    @Override
    public boolean isCancelled() {
      final SettableFuture<Boolean> retVal = SettableFuture.create();
      serializingExecutor.execute(new Runnable() {
        @Override
        public void run() {
          retVal.set(SerializingServerCall.super.isCancelled());
        }
      });
      try {
        return retVal.get();
      } catch (InterruptedException e) {
        throw new RuntimeException(ERROR_MSG, e);
      } catch (ExecutionException e) {
        throw new RuntimeException(ERROR_MSG, e);
      }
    }

    @Override
    public void setMessageCompression(final boolean enabled) {
      serializingExecutor.execute(new Runnable() {
        @Override
        public void run() {
          SerializingServerCall.super.setMessageCompression(enabled);
        }
      });
    }

    @Override
    @ExperimentalApi("https://github.com/grpc/grpc-java/issues/11021")
    public void setOnReadyThreshold(final int numBytes) {
      serializingExecutor.execute(new Runnable() {
        @Override
        public void run() {
          SerializingServerCall.super.setOnReadyThreshold(numBytes);
        }
      });
    }

    @Override
    public void setCompression(final String compressor) {
      serializingExecutor.execute(new Runnable() {
        @Override
        public void run() {
          SerializingServerCall.super.setCompression(compressor);
        }
      });
    }

    @Override
    public Attributes getAttributes() {
      final SettableFuture<Attributes> retVal = SettableFuture.create();
      serializingExecutor.execute(new Runnable() {
        @Override
        public void run() {
          retVal.set(SerializingServerCall.super.getAttributes());
        }
      });
      try {
        return retVal.get();
      } catch (InterruptedException e) {
        throw new RuntimeException(ERROR_MSG, e);
      } catch (ExecutionException e) {
        throw new RuntimeException(ERROR_MSG, e);
      }
    }

    @Nullable
    @Override
    public String getAuthority() {
      final SettableFuture<String> retVal = SettableFuture.create();
      serializingExecutor.execute(new Runnable() {
        @Override
        public void run() {
          retVal.set(SerializingServerCall.super.getAuthority());
        }
      });
      try {
        return retVal.get();
      } catch (InterruptedException e) {
        throw new RuntimeException(ERROR_MSG, e);
      } catch (ExecutionException e) {
        throw new RuntimeException(ERROR_MSG, e);
      }
    }

    @Override
    public void triggerEvent(final Object event) {
      serializingExecutor.execute(new Runnable() {
        @Override
        public void run() {
          SerializingServerCall.super.triggerEvent(event);
        }
      });
    }
  }
}

1	/*
2	* Copyright 2017 The gRPC Authors
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16
17	package io.grpc.util;
18
19	import com.google.common.util.concurrent.MoreExecutors;
20	import com.google.common.util.concurrent.SettableFuture;
21	import io.grpc.Attributes;
22	import io.grpc.ExperimentalApi;
23	import io.grpc.ForwardingServerCall;
24	import io.grpc.ForwardingServerCallListener;
25	import io.grpc.Metadata;
26	import io.grpc.ServerCall;
27	import io.grpc.ServerCallHandler;
28	import io.grpc.ServerInterceptor;
29	import io.grpc.Status;
30	import io.grpc.StatusRuntimeException;
31	import io.grpc.internal.SerializingExecutor;
32	import java.util.concurrent.ExecutionException;
33	import javax.annotation.Nullable;
34
35	/**
36	* A class that intercepts uncaught exceptions of type {@link StatusRuntimeException} and handles
37	* them by closing the {@link ServerCall}, and transmitting the exception's status and metadata
38	* to the client.
39	*
40	* <p>Without this interceptor, gRPC will strip all details and close the {@link ServerCall} with
41	* a generic {@link Status#UNKNOWN} code.
42	*
43	* <p>Security warning: the {@link Status} and {@link Metadata} may contain sensitive server-side
44	* state information, and generally should not be sent to clients. Only install this interceptor
45	* if all clients are trusted.
46	*/
47	@ExperimentalApi("https://github.com/grpc/grpc-java/issues/2189")
48	public final class TransmitStatusRuntimeExceptionInterceptor implements ServerInterceptor {
49	private TransmitStatusRuntimeExceptionInterceptor() {
50	}
51
52	public static ServerInterceptor instance() {
53	return new TransmitStatusRuntimeExceptionInterceptor();	1✔
54	}
55
56	@Override
57	public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
58	ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
59	final ServerCall<ReqT, RespT> serverCall = new SerializingServerCall<>(call);	1✔
60	ServerCall.Listener<ReqT> listener = next.startCall(serverCall, headers);	1✔
61	return new ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT>(listener) {	1✔
62	@Override
63	public void onMessage(ReqT message) {
64	try {
65	super.onMessage(message);	×
66	} catch (StatusRuntimeException e) {	1✔
67	closeWithException(e);	1✔
68	}	×
69	}	1✔
70
71	@Override
72	public void onHalfClose() {
73	try {
74	super.onHalfClose();	×
75	} catch (StatusRuntimeException e) {	1✔
76	closeWithException(e);	1✔
77	}	×
78	}	1✔
79
80	@Override
81	public void onCancel() {
82	try {
83	super.onCancel();	×
84	} catch (StatusRuntimeException e) {	1✔
85	closeWithException(e);	1✔
86	}	×
87	}	1✔
88
89	@Override
90	public void onComplete() {
91	try {
92	super.onComplete();	×
93	} catch (StatusRuntimeException e) {	1✔
94	closeWithException(e);	1✔
95	}	×
96	}	1✔
97
98	@Override
99	public void onReady() {
100	try {
101	super.onReady();	×
102	} catch (StatusRuntimeException e) {	1✔
103	closeWithException(e);	1✔
104	}	×
105	}	1✔
106
107	@Override
108	public void onEvent(Object event) {
109	try {
110	super.onEvent(event);	1✔
111	} catch (StatusRuntimeException e) {	1✔
112	closeWithException(e);	1✔
113	}	1✔
114	}	1✔
115
116	private void closeWithException(StatusRuntimeException t) {
117	Metadata metadata = t.getTrailers();	1✔
118	if (metadata == null) {	1✔
119	metadata = new Metadata();	×
120	}
121	serverCall.close(t.getStatus(), metadata);	1✔
122	}	1✔
123	};
124	}
125
126	/**
127	* A {@link ServerCall} that wraps around a non thread safe delegate and provides thread safe
128	* access by serializing everything on an executor.
129	*/
130	private static class SerializingServerCall<ReqT, RespT> extends
131	ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT> {
132	private static final String ERROR_MSG = "Encountered error during serialized access";
133	private final SerializingExecutor serializingExecutor =	1✔
134	new SerializingExecutor(MoreExecutors.directExecutor());	1✔
135	private boolean closeCalled = false;	1✔
136
137	SerializingServerCall(ServerCall<ReqT, RespT> delegate) {
138	super(delegate);	1✔
139	}	1✔
140
141	@Override
142	public void sendMessage(final RespT message) {
143	serializingExecutor.execute(new Runnable() {	1✔
144	@Override
145	public void run() {
146	SerializingServerCall.super.sendMessage(message);	1✔
147	}	1✔
148	});
149	}	1✔
150
151	@Override
152	public void request(final int numMessages) {
153	serializingExecutor.execute(new Runnable() {	×
154	@Override
155	public void run() {
156	SerializingServerCall.super.request(numMessages);	×
157	}	×
158	});
159	}	×
160
161	@Override
162	public void sendHeaders(final Metadata headers) {
163	serializingExecutor.execute(new Runnable() {	1✔
164	@Override
165	public void run() {
166	SerializingServerCall.super.sendHeaders(headers);	1✔
167	}	1✔
168	});
169	}	1✔
170
171	@Override
172	public void close(final Status status, final Metadata trailers) {
173	serializingExecutor.execute(new Runnable() {	1✔
174	@Override
175	public void run() {
176	if (!closeCalled) {	1✔
177	closeCalled = true;	1✔
178
179	SerializingServerCall.super.close(status, trailers);	1✔
180	}
181	}	1✔
182	});
183	}	1✔
184
185	@Override
186	public boolean isReady() {
187	final SettableFuture<Boolean> retVal = SettableFuture.create();	×
188	serializingExecutor.execute(new Runnable() {	×
189	@Override
190	public void run() {
191	retVal.set(SerializingServerCall.super.isReady());	×
192	}	×
193	});
194	try {
195	return retVal.get();	×
196	} catch (InterruptedException e) {	×
197	throw new RuntimeException(ERROR_MSG, e);	×
198	} catch (ExecutionException e) {	×
199	throw new RuntimeException(ERROR_MSG, e);	×
200	}
201	}
202
203	@Override
204	public boolean isCancelled() {
205	final SettableFuture<Boolean> retVal = SettableFuture.create();	×
206	serializingExecutor.execute(new Runnable() {	×
207	@Override
208	public void run() {
209	retVal.set(SerializingServerCall.super.isCancelled());	×
210	}	×
211	});
212	try {
213	return retVal.get();	×
214	} catch (InterruptedException e) {	×
215	throw new RuntimeException(ERROR_MSG, e);	×
216	} catch (ExecutionException e) {	×
217	throw new RuntimeException(ERROR_MSG, e);	×
218	}
219	}
220
221	@Override
222	public void setMessageCompression(final boolean enabled) {
223	serializingExecutor.execute(new Runnable() {	×
224	@Override
225	public void run() {
226	SerializingServerCall.super.setMessageCompression(enabled);	×
227	}	×
228	});
229	}	×
230
231	@Override
232	@ExperimentalApi("https://github.com/grpc/grpc-java/issues/11021")
233	public void setOnReadyThreshold(final int numBytes) {
234	serializingExecutor.execute(new Runnable() {	×
235	@Override
236	public void run() {
237	SerializingServerCall.super.setOnReadyThreshold(numBytes);	×
238	}	×
239	});
240	}	×
241
242	@Override
243	public void setCompression(final String compressor) {
244	serializingExecutor.execute(new Runnable() {	×
245	@Override
246	public void run() {
247	SerializingServerCall.super.setCompression(compressor);	×
248	}	×
249	});
250	}	×
251
252	@Override
253	public Attributes getAttributes() {
254	final SettableFuture<Attributes> retVal = SettableFuture.create();	×
255	serializingExecutor.execute(new Runnable() {	×
256	@Override
257	public void run() {
258	retVal.set(SerializingServerCall.super.getAttributes());	×
259	}	×
260	});
261	try {
262	return retVal.get();	×
263	} catch (InterruptedException e) {	×
264	throw new RuntimeException(ERROR_MSG, e);	×
265	} catch (ExecutionException e) {	×
266	throw new RuntimeException(ERROR_MSG, e);	×
267	}
268	}
269
270	@Nullable
271	@Override
272	public String getAuthority() {
273	final SettableFuture<String> retVal = SettableFuture.create();	×
274	serializingExecutor.execute(new Runnable() {	×
275	@Override
276	public void run() {
277	retVal.set(SerializingServerCall.super.getAuthority());	×
278	}	×
279	});
280	try {
281	return retVal.get();	×
282	} catch (InterruptedException e) {	×
283	throw new RuntimeException(ERROR_MSG, e);	×
284	} catch (ExecutionException e) {	×
285	throw new RuntimeException(ERROR_MSG, e);	×
286	}
287	}
288
289	@Override
290	public void triggerEvent(final Object event) {
291	serializingExecutor.execute(new Runnable() {	1✔
292	@Override
293	public void run() {
294	SerializingServerCall.super.triggerEvent(event);	1✔
295	}	1✔
296	});
297	}	1✔
298	}
299	}

grpc / grpc-java / #20469

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous