#18726

pending completion

Build # #18726

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Stabilize setExtensionRegistry() of ProtoLiteUtils and ProtoUtils (#10392)

Closes #1787

Run Details

29151 of 33044 relevant lines covered (88.22%)

0.88 hits per line

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

87.34

/../protobuf-lite/src/main/java/io/grpc/protobuf/lite/ProtoLiteUtils.java

/*
 * Copyright 2014 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.protobuf.lite;

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

import com.google.common.annotations.VisibleForTesting;
import com.google.protobuf.CodedInputStream;
import com.google.protobuf.ExtensionRegistryLite;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.MessageLite;
import com.google.protobuf.Parser;
import io.grpc.ExperimentalApi;
import io.grpc.KnownLength;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor.Marshaller;
import io.grpc.MethodDescriptor.PrototypeMarshaller;
import io.grpc.Status;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.lang.ref.Reference;
import java.lang.ref.WeakReference;

/**
 * Utility methods for using protobuf with grpc.
 */
@ExperimentalApi("Experimental until Lite is stable in protobuf")
public final class ProtoLiteUtils {

  // default visibility to avoid synthetic accessors
  static volatile ExtensionRegistryLite globalRegistry =
      ExtensionRegistryLite.getEmptyRegistry();

  private static final int BUF_SIZE = 8192;

  /**
   * The same value as {@link io.grpc.internal.GrpcUtil#DEFAULT_MAX_MESSAGE_SIZE}.
   */
  @VisibleForTesting
  static final int DEFAULT_MAX_MESSAGE_SIZE = 4 * 1024 * 1024;

  /**
   * Sets the global registry for proto marshalling shared across all servers and clients.
   *
   * <p>Warning:  This API will likely change over time.  It is not possible to have separate
   * registries per Process, Server, Channel, Service, or Method.  This is intentional until there
   * is a more appropriate API to set them.
   *
   * <p>Warning:  Do NOT modify the extension registry after setting it.  It is thread safe to call
   * {@link #setExtensionRegistry}, but not to modify the underlying object.
   *
   * <p>If you need custom parsing behavior for protos, you will need to make your own
   * {@code MethodDescriptor.Marshaller} for the time being.
   *
   * @since 1.0.0
   */
  public static void setExtensionRegistry(ExtensionRegistryLite newRegistry) {
    globalRegistry = checkNotNull(newRegistry, "newRegistry");
  }

  /**
   * Creates a {@link Marshaller} for protos of the same type as {@code defaultInstance}.
   *
   * @since 1.0.0
   */
  public static <T extends MessageLite> Marshaller<T> marshaller(T defaultInstance) {
    // TODO(ejona): consider changing return type to PrototypeMarshaller (assuming ABI safe)
    return new MessageMarshaller<>(defaultInstance, -1);
  }

  /**
   * Creates a {@link Marshaller} for protos of the same type as {@code defaultInstance} and a
   * custom limit for the recursion depth. Any negative number will leave the limit to its default
   * value as defined by the protobuf library.
   *
   * @since 1.56.0
   */
  @ExperimentalApi("https://github.com/grpc/grpc-java/issues/10108")
  public static <T extends MessageLite> Marshaller<T> marshallerWithRecursionLimit(
      T defaultInstance, int recursionLimit) {
    return new MessageMarshaller<>(defaultInstance, recursionLimit);
  }

  /**
   * Produce a metadata marshaller for a protobuf type.
   *
   * @since 1.0.0
   */
  public static <T extends MessageLite> Metadata.BinaryMarshaller<T> metadataMarshaller(
      T defaultInstance) {
    return new MetadataMarshaller<>(defaultInstance);
  }

  /** Copies the data from input stream to output stream. */
  static long copy(InputStream from, OutputStream to) throws IOException {
    // Copied from guava com.google.common.io.ByteStreams because its API is unstable (beta)
    checkNotNull(from, "inputStream cannot be null!");
    checkNotNull(to, "outputStream cannot be null!");
    byte[] buf = new byte[BUF_SIZE];
    long total = 0;
    while (true) {
      int r = from.read(buf);
      if (r == -1) {
        break;
      }
      to.write(buf, 0, r);
      total += r;
    }
    return total;
  }

  private ProtoLiteUtils() {
  }

  private static final class MessageMarshaller<T extends MessageLite>
      implements PrototypeMarshaller<T> {

    private static final ThreadLocal<Reference<byte[]>> bufs = new ThreadLocal<>();

    private final Parser<T> parser;
    private final T defaultInstance;
    private final int recursionLimit;

    @SuppressWarnings("unchecked")
    MessageMarshaller(T defaultInstance, int recursionLimit) {
      this.defaultInstance = checkNotNull(defaultInstance, "defaultInstance cannot be null");
      this.parser = (Parser<T>) defaultInstance.getParserForType();
      this.recursionLimit = recursionLimit;
    }

    @SuppressWarnings("unchecked")
    @Override
    public Class<T> getMessageClass() {
      // Precisely T since protobuf doesn't let messages extend other messages.
      return (Class<T>) defaultInstance.getClass();
    }

    @Override
    public T getMessagePrototype() {
      return defaultInstance;
    }

    @Override
    public InputStream stream(T value) {
      return new ProtoInputStream(value, parser);
    }

    @Override
    public T parse(InputStream stream) {
      if (stream instanceof ProtoInputStream) {
        ProtoInputStream protoStream = (ProtoInputStream) stream;
        // Optimization for in-memory transport. Returning provided object is safe since protobufs
        // are immutable.
        //
        // However, we can't assume the types match, so we have to verify the parser matches.
        // Today the parser is always the same for a given proto, but that isn't guaranteed. Even
        // if not, using the same MethodDescriptor would ensure the parser matches and permit us
        // to enable this optimization.
        if (protoStream.parser() == parser) {
          try {
            @SuppressWarnings("unchecked")
            T message = (T) ((ProtoInputStream) stream).message();
            return message;
          } catch (IllegalStateException ignored) {
            // Stream must have been read from, which is a strange state. Since the point of this
            // optimization is to be transparent, instead of throwing an error we'll continue,
            // even though it seems likely there's a bug.
          }
        }
      }
      CodedInputStream cis = null;
      try {
        if (stream instanceof KnownLength) {
          int size = stream.available();
          if (size > 0 && size <= DEFAULT_MAX_MESSAGE_SIZE) {
            Reference<byte[]> ref;
            // buf should not be used after this method has returned.
            byte[] buf;
            if ((ref = bufs.get()) == null || (buf = ref.get()) == null || buf.length < size) {
              buf = new byte[size];
              bufs.set(new WeakReference<>(buf));
            }

            int remaining = size;
            while (remaining > 0) {
              int position = size - remaining;
              int count = stream.read(buf, position, remaining);
              if (count == -1) {
                break;
              }
              remaining -= count;
            }

            if (remaining != 0) {
              int position = size - remaining;
              throw new RuntimeException("size inaccurate: " + size + " != " + position);
            }
            cis = CodedInputStream.newInstance(buf, 0, size);
          } else if (size == 0) {
            return defaultInstance;
          }
        }
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
      if (cis == null) {
        cis = CodedInputStream.newInstance(stream);
      }
      // Pre-create the CodedInputStream so that we can remove the size limit restriction
      // when parsing.
      cis.setSizeLimit(Integer.MAX_VALUE);

      if (recursionLimit >= 0) {
        cis.setRecursionLimit(recursionLimit);
      }

      try {
        return parseFrom(cis);
      } catch (InvalidProtocolBufferException ipbe) {
        throw Status.INTERNAL.withDescription("Invalid protobuf byte sequence")
            .withCause(ipbe).asRuntimeException();
      }
    }

    private T parseFrom(CodedInputStream stream) throws InvalidProtocolBufferException {
      T message = parser.parseFrom(stream, globalRegistry);
      try {
        stream.checkLastTagWas(0);
        return message;
      } catch (InvalidProtocolBufferException e) {
        e.setUnfinishedMessage(message);
        throw e;
      }
    }
  }

  private static final class MetadataMarshaller<T extends MessageLite>
      implements Metadata.BinaryMarshaller<T> {

    private final T defaultInstance;

    MetadataMarshaller(T defaultInstance) {
      this.defaultInstance = defaultInstance;
    }

    @Override
    public byte[] toBytes(T value) {
      return value.toByteArray();
    }

    @Override
    @SuppressWarnings("unchecked")
    public T parseBytes(byte[] serialized) {
      try {
        return (T) defaultInstance.getParserForType().parseFrom(serialized, globalRegistry);
      } catch (InvalidProtocolBufferException ipbe) {
        throw new IllegalArgumentException(ipbe);
      }
    }
  }
}

1	/*
2	* Copyright 2014 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.protobuf.lite;
18
19	import static com.google.common.base.Preconditions.checkNotNull;
20
21	import com.google.common.annotations.VisibleForTesting;
22	import com.google.protobuf.CodedInputStream;
23	import com.google.protobuf.ExtensionRegistryLite;
24	import com.google.protobuf.InvalidProtocolBufferException;
25	import com.google.protobuf.MessageLite;
26	import com.google.protobuf.Parser;
27	import io.grpc.ExperimentalApi;
28	import io.grpc.KnownLength;
29	import io.grpc.Metadata;
30	import io.grpc.MethodDescriptor.Marshaller;
31	import io.grpc.MethodDescriptor.PrototypeMarshaller;
32	import io.grpc.Status;
33	import java.io.IOException;
34	import java.io.InputStream;
35	import java.io.OutputStream;
36	import java.lang.ref.Reference;
37	import java.lang.ref.WeakReference;
38
39	/**
40	* Utility methods for using protobuf with grpc.
41	*/
42	@ExperimentalApi("Experimental until Lite is stable in protobuf")
43	public final class ProtoLiteUtils {
44
45	// default visibility to avoid synthetic accessors
46	static volatile ExtensionRegistryLite globalRegistry =	1✔
47	ExtensionRegistryLite.getEmptyRegistry();	1✔
48
49	private static final int BUF_SIZE = 8192;
50
51	/**
52	* The same value as {@link io.grpc.internal.GrpcUtil#DEFAULT_MAX_MESSAGE_SIZE}.
53	*/
54	@VisibleForTesting
55	static final int DEFAULT_MAX_MESSAGE_SIZE = 4 * 1024 * 1024;
56
57	/**
58	* Sets the global registry for proto marshalling shared across all servers and clients.
59	*
60	* <p>Warning: This API will likely change over time. It is not possible to have separate
61	* registries per Process, Server, Channel, Service, or Method. This is intentional until there
62	* is a more appropriate API to set them.
63	*
64	* <p>Warning: Do NOT modify the extension registry after setting it. It is thread safe to call
65	* {@link #setExtensionRegistry}, but not to modify the underlying object.
66	*
67	* <p>If you need custom parsing behavior for protos, you will need to make your own
68	* {@code MethodDescriptor.Marshaller} for the time being.
69	*
70	* @since 1.0.0
71	*/
72	public static void setExtensionRegistry(ExtensionRegistryLite newRegistry) {
73	globalRegistry = checkNotNull(newRegistry, "newRegistry");	×
74	}	×
75
76	/**
77	* Creates a {@link Marshaller} for protos of the same type as {@code defaultInstance}.
78	*
79	* @since 1.0.0
80	*/
81	public static <T extends MessageLite> Marshaller<T> marshaller(T defaultInstance) {
82	// TODO(ejona): consider changing return type to PrototypeMarshaller (assuming ABI safe)
83	return new MessageMarshaller<>(defaultInstance, -1);	1✔
84	}
85
86	/**
87	* Creates a {@link Marshaller} for protos of the same type as {@code defaultInstance} and a
88	* custom limit for the recursion depth. Any negative number will leave the limit to its default
89	* value as defined by the protobuf library.
90	*
91	* @since 1.56.0
92	*/
93	@ExperimentalApi("https://github.com/grpc/grpc-java/issues/10108")
94	public static <T extends MessageLite> Marshaller<T> marshallerWithRecursionLimit(
95	T defaultInstance, int recursionLimit) {
96	return new MessageMarshaller<>(defaultInstance, recursionLimit);	1✔
97	}
98
99	/**
100	* Produce a metadata marshaller for a protobuf type.
101	*
102	* @since 1.0.0
103	*/
104	public static <T extends MessageLite> Metadata.BinaryMarshaller<T> metadataMarshaller(
105	T defaultInstance) {
106	return new MetadataMarshaller<>(defaultInstance);	1✔
107	}
108
109	/** Copies the data from input stream to output stream. */
110	static long copy(InputStream from, OutputStream to) throws IOException {
111	// Copied from guava com.google.common.io.ByteStreams because its API is unstable (beta)
112	checkNotNull(from, "inputStream cannot be null!");	1✔
113	checkNotNull(to, "outputStream cannot be null!");	1✔
114	byte[] buf = new byte[BUF_SIZE];	1✔
115	long total = 0;	1✔
116	while (true) {
117	int r = from.read(buf);	1✔
118	if (r == -1) {	1✔
119	break;	1✔
120	}
121	to.write(buf, 0, r);	1✔
122	total += r;	1✔
123	}	1✔
124	return total;	1✔
125	}
126
127	private ProtoLiteUtils() {
128	}
129
130	private static final class MessageMarshaller<T extends MessageLite>
131	implements PrototypeMarshaller<T> {
132
133	private static final ThreadLocal<Reference<byte[]>> bufs = new ThreadLocal<>();	1✔
134
135	private final Parser<T> parser;
136	private final T defaultInstance;
137	private final int recursionLimit;
138
139	@SuppressWarnings("unchecked")
140	MessageMarshaller(T defaultInstance, int recursionLimit) {	1✔
141	this.defaultInstance = checkNotNull(defaultInstance, "defaultInstance cannot be null");	1✔
142	this.parser = (Parser<T>) defaultInstance.getParserForType();	1✔
143	this.recursionLimit = recursionLimit;	1✔
144	}	1✔
145
146	@SuppressWarnings("unchecked")
147	@Override
148	public Class<T> getMessageClass() {
149	// Precisely T since protobuf doesn't let messages extend other messages.
150	return (Class<T>) defaultInstance.getClass();	1✔
151	}
152
153	@Override
154	public T getMessagePrototype() {
155	return defaultInstance;	1✔
156	}
157
158	@Override
159	public InputStream stream(T value) {
160	return new ProtoInputStream(value, parser);	1✔
161	}
162
163	@Override
164	public T parse(InputStream stream) {
165	if (stream instanceof ProtoInputStream) {	1✔
166	ProtoInputStream protoStream = (ProtoInputStream) stream;	1✔
167	// Optimization for in-memory transport. Returning provided object is safe since protobufs
168	// are immutable.
169	//
170	// However, we can't assume the types match, so we have to verify the parser matches.
171	// Today the parser is always the same for a given proto, but that isn't guaranteed. Even
172	// if not, using the same MethodDescriptor would ensure the parser matches and permit us
173	// to enable this optimization.
174	if (protoStream.parser() == parser) {	1✔
175	try {
176	@SuppressWarnings("unchecked")
177	T message = (T) ((ProtoInputStream) stream).message();	1✔
178	return message;	1✔
179	} catch (IllegalStateException ignored) {	1✔
180	// Stream must have been read from, which is a strange state. Since the point of this
181	// optimization is to be transparent, instead of throwing an error we'll continue,
182	// even though it seems likely there's a bug.
183	}
184	}
185	}
186	CodedInputStream cis = null;	1✔
187	try {
188	if (stream instanceof KnownLength) {	1✔
189	int size = stream.available();	1✔
190	if (size > 0 && size <= DEFAULT_MAX_MESSAGE_SIZE) {	1✔
191	Reference<byte[]> ref;
192	// buf should not be used after this method has returned.
193	byte[] buf;
194	if ((ref = bufs.get()) == null \|\| (buf = ref.get()) == null \|\| buf.length < size) {	1✔
195	buf = new byte[size];	1✔
196	bufs.set(new WeakReference<>(buf));	1✔
197	}
198
199	int remaining = size;	1✔
200	while (remaining > 0) {	1✔
201	int position = size - remaining;	1✔
202	int count = stream.read(buf, position, remaining);	1✔
203	if (count == -1) {	1✔
204	break;	×
205	}
206	remaining -= count;	1✔
207	}	1✔
208
209	if (remaining != 0) {	1✔
210	int position = size - remaining;	×
211	throw new RuntimeException("size inaccurate: " + size + " != " + position);	×
212	}
213	cis = CodedInputStream.newInstance(buf, 0, size);	1✔
214	} else if (size == 0) {	1✔
215	return defaultInstance;	1✔
216	}
217	}
218	} catch (IOException e) {	×
219	throw new RuntimeException(e);	×
220	}	1✔
221	if (cis == null) {	1✔
222	cis = CodedInputStream.newInstance(stream);	1✔
223	}
224	// Pre-create the CodedInputStream so that we can remove the size limit restriction
225	// when parsing.
226	cis.setSizeLimit(Integer.MAX_VALUE);	1✔
227
228	if (recursionLimit >= 0) {	1✔
229	cis.setRecursionLimit(recursionLimit);	1✔
230	}
231
232	try {
233	return parseFrom(cis);	1✔
234	} catch (InvalidProtocolBufferException ipbe) {	1✔
235	throw Status.INTERNAL.withDescription("Invalid protobuf byte sequence")	1✔
236	.withCause(ipbe).asRuntimeException();	1✔
237	}
238	}
239
240	private T parseFrom(CodedInputStream stream) throws InvalidProtocolBufferException {
241	T message = parser.parseFrom(stream, globalRegistry);	1✔
242	try {
243	stream.checkLastTagWas(0);	1✔
244	return message;	1✔
245	} catch (InvalidProtocolBufferException e) {	×
246	e.setUnfinishedMessage(message);	×
247	throw e;	×
248	}
249	}
250	}
251
252	private static final class MetadataMarshaller<T extends MessageLite>
253	implements Metadata.BinaryMarshaller<T> {
254
255	private final T defaultInstance;
256
257	MetadataMarshaller(T defaultInstance) {	1✔
258	this.defaultInstance = defaultInstance;	1✔
259	}	1✔
260
261	@Override
262	public byte[] toBytes(T value) {
263	return value.toByteArray();	1✔
264	}
265
266	@Override
267	@SuppressWarnings("unchecked")
268	public T parseBytes(byte[] serialized) {
269	try {
270	return (T) defaultInstance.getParserForType().parseFrom(serialized, globalRegistry);	1✔
271	} catch (InvalidProtocolBufferException ipbe) {	1✔
272	throw new IllegalArgumentException(ipbe);	1✔
273	}
274	}
275	}
276	}

grpc / grpc-java / #18726

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