#20429

Committed 01 Sep 2026 09:05AM UTC coverage: 89.215% (+0.04%) from 89.174%

Build # #20429

Build Type

push

github

Committed by

web-flow

Commit Message

xds: Ext proc client application level flow control (#12956)

Coverage Stats

38638 of 43309 relevant lines covered (89.21%)

0.89 hits per line

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

81.05

/../okhttp/src/main/java/io/grpc/okhttp/ExceptionHandlingFrameWriter.java

/*
 * Copyright 2018 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.okhttp;

import static com.google.common.base.Preconditions.checkNotNull;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import io.grpc.okhttp.internal.framed.ErrorCode;
import io.grpc.okhttp.internal.framed.FrameWriter;
import io.grpc.okhttp.internal.framed.Header;
import io.grpc.okhttp.internal.framed.Settings;
import java.io.IOException;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;
import okio.Buffer;
import okio.ByteString;

/**
 * FrameWriter that propagates IOExceptions via callback instead of throwing. This allows
 * centralized handling of errors. Exceptions only impact the single call that throws them; callers
 * should be sure to kill the connection after an exception (potentially after sending a GOAWAY) as
 * otherwise additional frames after the failed/omitted one could cause HTTP/2 confusion.
 */
final class ExceptionHandlingFrameWriter implements FrameWriter {

  private static final Logger log = Logger.getLogger(OkHttpClientTransport.class.getName());

  private final TransportExceptionHandler transportExceptionHandler;

  private final FrameWriter frameWriter;

  private final OkHttpFrameLogger frameLogger =
      new OkHttpFrameLogger(Level.FINE, OkHttpClientTransport.class);

  ExceptionHandlingFrameWriter(
      TransportExceptionHandler transportExceptionHandler, FrameWriter frameWriter) {
    this.transportExceptionHandler =
        checkNotNull(transportExceptionHandler, "transportExceptionHandler");
    this.frameWriter = Preconditions.checkNotNull(frameWriter, "frameWriter");
  }

  @Override
  public void connectionPreface() {
    try {
      frameWriter.connectionPreface();
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void ackSettings(Settings peerSettings) {
    frameLogger.logSettingsAck(OkHttpFrameLogger.Direction.OUTBOUND);
    try {
      frameWriter.ackSettings(peerSettings);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void pushPromise(int streamId, int promisedStreamId, List<Header> requestHeaders) {
    frameLogger.logPushPromise(OkHttpFrameLogger.Direction.OUTBOUND,
        streamId, promisedStreamId, requestHeaders);
    try {
      frameWriter.pushPromise(streamId, promisedStreamId, requestHeaders);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void flush() {
    try {
      frameWriter.flush();
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void synStream(
      boolean outFinished,
      boolean inFinished,
      int streamId,
      int associatedStreamId,
      List<Header> headerBlock) {
    try {
      frameWriter.synStream(outFinished, inFinished, streamId, associatedStreamId, headerBlock);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void synReply(boolean outFinished, int streamId,
      List<Header> headerBlock) {
    try {
      frameWriter.synReply(outFinished, streamId, headerBlock);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void headers(int streamId, List<Header> headerBlock) {
    frameLogger.logHeaders(OkHttpFrameLogger.Direction.OUTBOUND, streamId, headerBlock, false);
    try {
      frameWriter.headers(streamId, headerBlock);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void rstStream(int streamId, ErrorCode errorCode) {
    frameLogger.logRstStream(OkHttpFrameLogger.Direction.OUTBOUND, streamId, errorCode);
    try {
      frameWriter.rstStream(streamId, errorCode);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public int maxDataLength() {
    return frameWriter.maxDataLength();
  }

  @Override
  public void data(boolean outFinished, int streamId, Buffer source, int byteCount) {
    frameLogger.logData(OkHttpFrameLogger.Direction.OUTBOUND,
        streamId, source.buffer(), byteCount, outFinished);
    try {
      frameWriter.data(outFinished, streamId, source, byteCount);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void settings(Settings okHttpSettings) {
    frameLogger.logSettings(OkHttpFrameLogger.Direction.OUTBOUND, okHttpSettings);
    try {
      frameWriter.settings(okHttpSettings);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void ping(boolean ack, int payload1, int payload2) {
    if (ack) {
      frameLogger.logPingAck(OkHttpFrameLogger.Direction.OUTBOUND,
          ((long) payload1 << 32) | (payload2 & 0xFFFFFFFFL));
    } else {
      frameLogger.logPing(OkHttpFrameLogger.Direction.OUTBOUND,
          ((long) payload1 << 32) | (payload2 & 0xFFFFFFFFL));
    }
    try {
      frameWriter.ping(ack, payload1, payload2);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void goAway(int lastGoodStreamId, ErrorCode errorCode,
      byte[] debugData) {
    frameLogger.logGoAway(OkHttpFrameLogger.Direction.OUTBOUND,
        lastGoodStreamId, errorCode, ByteString.of(debugData));
    try {
      frameWriter.goAway(lastGoodStreamId, errorCode, debugData);
      // Flush it since after goAway, we are likely to close this writer.
      frameWriter.flush();
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void windowUpdate(int streamId, long windowSizeIncrement) {
    frameLogger.logWindowsUpdate(OkHttpFrameLogger.Direction.OUTBOUND,
        streamId, windowSizeIncrement);
    try {
      frameWriter.windowUpdate(streamId, windowSizeIncrement);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void close() {
    try {
      frameWriter.close();
    } catch (IOException e) {
      log.log(getLogLevel(e), "Failed closing connection", e);
    }
  }

  /**
   * Accepts a throwable and returns the appropriate logging level. Uninteresting exceptions
   * should not clutter the log.
   */
  @VisibleForTesting
  static Level getLogLevel(Throwable t) {
    if (t.getClass().equals(IOException.class)) {
      return Level.FINE;
    }
    return Level.INFO;
  }

  /** A class that handles transport exception. */
  interface TransportExceptionHandler {
    /** Handles exception. */
    void onException(Throwable throwable);
  }
}

1	/*
2	* Copyright 2018 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.okhttp;
18
19	import static com.google.common.base.Preconditions.checkNotNull;
20
21	import com.google.common.annotations.VisibleForTesting;
22	import com.google.common.base.Preconditions;
23	import io.grpc.okhttp.internal.framed.ErrorCode;
24	import io.grpc.okhttp.internal.framed.FrameWriter;
25	import io.grpc.okhttp.internal.framed.Header;
26	import io.grpc.okhttp.internal.framed.Settings;
27	import java.io.IOException;
28	import java.util.List;
29	import java.util.logging.Level;
30	import java.util.logging.Logger;
31	import okio.Buffer;
32	import okio.ByteString;
33
34	/**
35	* FrameWriter that propagates IOExceptions via callback instead of throwing. This allows
36	* centralized handling of errors. Exceptions only impact the single call that throws them; callers
37	* should be sure to kill the connection after an exception (potentially after sending a GOAWAY) as
38	* otherwise additional frames after the failed/omitted one could cause HTTP/2 confusion.
39	*/
40	final class ExceptionHandlingFrameWriter implements FrameWriter {
41
42	private static final Logger log = Logger.getLogger(OkHttpClientTransport.class.getName());	1 ✔
43
44	private final TransportExceptionHandler transportExceptionHandler;
45
46	private final FrameWriter frameWriter;
47
48	private final OkHttpFrameLogger frameLogger =	1 ✔
49	new OkHttpFrameLogger(Level.FINE, OkHttpClientTransport.class);
50
51	ExceptionHandlingFrameWriter(
52	TransportExceptionHandler transportExceptionHandler, FrameWriter frameWriter) {	1 ✔
53	this.transportExceptionHandler =	1 ✔
54	checkNotNull(transportExceptionHandler, "transportExceptionHandler");	1 ✔
55	this.frameWriter = Preconditions.checkNotNull(frameWriter, "frameWriter");	1 ✔
56	}	1 ✔
57
58	@Override
59	public void connectionPreface() {
60	try {
61	frameWriter.connectionPreface();	1 ✔
62	} catch (IOException e) {	×
63	transportExceptionHandler.onException(e);	×
64	}	1 ✔
65	}	1 ✔
66
67	@Override
68	public void ackSettings(Settings peerSettings) {
69	frameLogger.logSettingsAck(OkHttpFrameLogger.Direction.OUTBOUND);	1 ✔
70	try {
71	frameWriter.ackSettings(peerSettings);	1 ✔
72	} catch (IOException e) {	×
73	transportExceptionHandler.onException(e);	×
74	}	1 ✔
75	}	1 ✔
76
77	@Override
78	public void pushPromise(int streamId, int promisedStreamId, List<Header> requestHeaders) {
79	frameLogger.logPushPromise(OkHttpFrameLogger.Direction.OUTBOUND,	1 ✔
80	streamId, promisedStreamId, requestHeaders);
81	try {
82	frameWriter.pushPromise(streamId, promisedStreamId, requestHeaders);	1 ✔
83	} catch (IOException e) {	×
84	transportExceptionHandler.onException(e);	×
85	}	1 ✔
86	}	1 ✔
87
88	@Override
89	public void flush() {
90	try {
91	frameWriter.flush();	1 ✔
92	} catch (IOException e) {	1 ✔
93	transportExceptionHandler.onException(e);	1 ✔
94	}	1 ✔
95	}	1 ✔
96
97	@Override
98	public void synStream(
99	boolean outFinished,
100	boolean inFinished,
101	int streamId,
102	int associatedStreamId,
103	List<Header> headerBlock) {
104	try {
105	frameWriter.synStream(outFinished, inFinished, streamId, associatedStreamId, headerBlock);	1 ✔
106	} catch (IOException e) {	×
107	transportExceptionHandler.onException(e);	×
108	}	1 ✔
109	}	1 ✔
110
111	@Override
112	public void synReply(boolean outFinished, int streamId,
113	List<Header> headerBlock) {
114	try {
115	frameWriter.synReply(outFinished, streamId, headerBlock);	1 ✔
116	} catch (IOException e) {	1 ✔
117	transportExceptionHandler.onException(e);	1 ✔
118	}	1 ✔
119	}	1 ✔
120
121	@Override
122	public void headers(int streamId, List<Header> headerBlock) {
123	frameLogger.logHeaders(OkHttpFrameLogger.Direction.OUTBOUND, streamId, headerBlock, false);	1 ✔
124	try {
125	frameWriter.headers(streamId, headerBlock);	1 ✔
126	} catch (IOException e) {	×
127	transportExceptionHandler.onException(e);	×
128	}	1 ✔
129	}	1 ✔
130
131	@Override
132	public void rstStream(int streamId, ErrorCode errorCode) {
133	frameLogger.logRstStream(OkHttpFrameLogger.Direction.OUTBOUND, streamId, errorCode);	1 ✔
134	try {
135	frameWriter.rstStream(streamId, errorCode);	1 ✔
136	} catch (IOException e) {	1 ✔
137	transportExceptionHandler.onException(e);	1 ✔
138	}	1 ✔
139	}	1 ✔
140
141	@Override
142	public int maxDataLength() {
143	return frameWriter.maxDataLength();	1 ✔
144	}
145
146	@Override
147	public void data(boolean outFinished, int streamId, Buffer source, int byteCount) {
148	frameLogger.logData(OkHttpFrameLogger.Direction.OUTBOUND,	1 ✔
149	streamId, source.buffer(), byteCount, outFinished);	1 ✔
150	try {
151	frameWriter.data(outFinished, streamId, source, byteCount);	1 ✔
152	} catch (IOException e) {	×
153	transportExceptionHandler.onException(e);	×
154	}	1 ✔
155	}	1 ✔
156
157	@Override
158	public void settings(Settings okHttpSettings) {
159	frameLogger.logSettings(OkHttpFrameLogger.Direction.OUTBOUND, okHttpSettings);	1 ✔
160	try {
161	frameWriter.settings(okHttpSettings);	1 ✔
162	} catch (IOException e) {	×
163	transportExceptionHandler.onException(e);	×
164	}	1 ✔
165	}	1 ✔
166
167	@Override
168	public void ping(boolean ack, int payload1, int payload2) {
169	if (ack) {	1 ✔
170	frameLogger.logPingAck(OkHttpFrameLogger.Direction.OUTBOUND,	1 ✔
171	((long) payload1 << 32) \| (payload2 & 0xFFFFFFFFL));
172	} else {
173	frameLogger.logPing(OkHttpFrameLogger.Direction.OUTBOUND,	1 ✔
174	((long) payload1 << 32) \| (payload2 & 0xFFFFFFFFL));
175	}
176	try {
177	frameWriter.ping(ack, payload1, payload2);	1 ✔
178	} catch (IOException e) {	1 ✔
179	transportExceptionHandler.onException(e);	1 ✔
180	}	1 ✔
181	}	1 ✔
182
183	@Override
184	public void goAway(int lastGoodStreamId, ErrorCode errorCode,
185	byte[] debugData) {
186	frameLogger.logGoAway(OkHttpFrameLogger.Direction.OUTBOUND,	1 ✔
187	lastGoodStreamId, errorCode, ByteString.of(debugData));	1 ✔
188	try {
189	frameWriter.goAway(lastGoodStreamId, errorCode, debugData);	1 ✔
190	// Flush it since after goAway, we are likely to close this writer.
191	frameWriter.flush();	1 ✔
192	} catch (IOException e) {	1 ✔
193	transportExceptionHandler.onException(e);	1 ✔
194	}	1 ✔
195	}	1 ✔
196
197	@Override
198	public void windowUpdate(int streamId, long windowSizeIncrement) {
199	frameLogger.logWindowsUpdate(OkHttpFrameLogger.Direction.OUTBOUND,	1 ✔
200	streamId, windowSizeIncrement);
201	try {
202	frameWriter.windowUpdate(streamId, windowSizeIncrement);	1 ✔
203	} catch (IOException e) {	×
204	transportExceptionHandler.onException(e);	×
205	}	1 ✔
206	}	1 ✔
207
208	@Override
209	public void close() {
210	try {
211	frameWriter.close();	1 ✔
212	} catch (IOException e) {	×
213	log.log(getLogLevel(e), "Failed closing connection", e);	×
214	}	1 ✔
215	}	1 ✔
216
217	/**
218	* Accepts a throwable and returns the appropriate logging level. Uninteresting exceptions
219	* should not clutter the log.
220	*/
221	@VisibleForTesting
222	static Level getLogLevel(Throwable t) {
223	if (t.getClass().equals(IOException.class)) {	1 ✔
224	return Level.FINE;	1 ✔
225	}
226	return Level.INFO;	1 ✔
227	}
228
229	/** A class that handles transport exception. */
230	interface TransportExceptionHandler {
231	/** Handles exception. */
232	void onException(Throwable throwable);
233	}
234	}

grpc / grpc-java / #20429

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