|
19 | 19 | import com.google.api.gax.rpc.ApiCallContext; |
20 | 20 | import com.google.api.gax.rpc.ApiExceptionFactory; |
21 | 21 | import com.google.api.gax.rpc.ApiStreamObserver; |
22 | | -import com.google.api.gax.rpc.BidiStreamingCallable; |
23 | 22 | import com.google.api.gax.rpc.ClientStreamingCallable; |
24 | 23 | import com.google.api.gax.rpc.ResponseObserver; |
25 | 24 | import com.google.api.gax.rpc.ServerStreamingCallable; |
|
36 | 35 |
|
37 | 36 | @InternalApi("for testing") |
38 | 37 | public class FakeStreamingApi { |
39 | | - |
40 | | - public static class BidiStreamingStashCallable<RequestT, ResponseT> |
41 | | - extends BidiStreamingCallable<RequestT, ResponseT> { |
42 | | - private ApiCallContext context; |
43 | | - private ApiStreamObserver<ResponseT> responseObserver; |
44 | | - private AccumulatingStreamObserver<RequestT> requestObserver; |
45 | | - private List<ResponseT> responseList; |
46 | | - |
47 | | - public BidiStreamingStashCallable() { |
48 | | - responseList = new ArrayList<>(); |
49 | | - } |
50 | | - |
51 | | - public BidiStreamingStashCallable(List<ResponseT> responseList) { |
52 | | - this.responseList = responseList; |
53 | | - } |
54 | | - |
55 | | - @Override |
56 | | - public ApiStreamObserver<RequestT> bidiStreamingCall( |
57 | | - ApiStreamObserver<ResponseT> responseObserver, ApiCallContext context) { |
58 | | - Preconditions.checkNotNull(responseObserver); |
59 | | - this.responseObserver = responseObserver; |
60 | | - this.context = context; |
61 | | - this.requestObserver = new AccumulatingStreamObserver<>(); |
62 | | - return requestObserver; |
63 | | - } |
64 | | - |
65 | | - public ApiCallContext getContext() { |
66 | | - return context; |
67 | | - } |
68 | | - |
69 | | - public ApiStreamObserver<ResponseT> getActualObserver() { |
70 | | - return responseObserver; |
71 | | - } |
72 | | - |
73 | | - public List<RequestT> getActualRequests() { |
74 | | - return requestObserver.getValues(); |
75 | | - } |
76 | | - |
77 | | - private void sendResponses() { |
78 | | - for (ResponseT response : responseList) { |
79 | | - responseObserver.onNext(response); |
80 | | - } |
81 | | - responseObserver.onCompleted(); |
82 | | - } |
83 | | - |
84 | | - private class AccumulatingStreamObserver<T> implements ApiStreamObserver<T> { |
85 | | - private List<T> requestList = new ArrayList<>(); |
86 | | - private Throwable error; |
87 | | - private boolean completed = false; |
88 | | - |
89 | | - @Override |
90 | | - public void onNext(T value) { |
91 | | - requestList.add(value); |
92 | | - } |
93 | | - |
94 | | - @Override |
95 | | - public void onError(Throwable t) { |
96 | | - error = t; |
97 | | - } |
98 | | - |
99 | | - @Override |
100 | | - public void onCompleted() { |
101 | | - completed = true; |
102 | | - BidiStreamingStashCallable.this.sendResponses(); |
103 | | - } |
104 | | - |
105 | | - public List<T> getValues() { |
106 | | - if (!completed) { |
107 | | - throw new IllegalStateException("Stream not completed."); |
108 | | - } |
109 | | - if (error != null) { |
110 | | - throw ApiExceptionFactory.createException(error, FakeStatusCode.of(Code.UNKNOWN), false); |
111 | | - } |
112 | | - return requestList; |
113 | | - } |
114 | | - } |
115 | | - } |
116 | | - |
117 | 38 | public static class ServerStreamingStashCallable<RequestT, ResponseT> |
118 | 39 | extends ServerStreamingCallable<RequestT, ResponseT> { |
119 | 40 | private ApiCallContext context; |
|
0 commit comments