JevTransport.java
原始文件
1package com.android.commands.monkey.ape.jev;
2
3import java.io.*;
4import java.net.*;
5import java.nio.charset.StandardCharsets;
6import java.util.Timer;
7import java.util.TimerTask;
8import java.util.concurrent.ArrayBlockingQueue;
9import java.util.concurrent.TimeUnit;
10import java.util.Map;
11import java.util.Collections;
12import java.util.LinkedHashMap;
13
14/** One bounded HTTP worker. Deadlines include the queue and never wait for socket cleanup. */
15public final class JevTransport implements Closeable {
16 interface Completion { void completed(Result result); }
17 public static final class Result {
18 public final String raw,error;
19 public final int httpStatus;
20 public final long submittedNanos,sentNanos,completedNanos;
21 private Result(Call call,String raw,String error,int status) {
22 this.raw=raw;this.error=error;httpStatus=status;submittedNanos=call==null?System.nanoTime():call.submitted;
23 sentNanos=call==null?submittedNanos:call.sent;completedNanos=System.nanoTime();
24 }
25 static Result cached(String raw){return new Result(null,raw,null,200);}
26 public long waitNanos(){return completedNanos-submittedNanos;}
27 public long latencyNanos(){return sentNanos==0?0:completedNanos-sentNanos;}
28 }
29 public final class Call {
30 public final String requestId;
31 private final byte[] body;
32 private final int responseLimit;
33 private final long submitted=System.nanoTime(),deadline;
34 private volatile long sent;
35 private volatile HttpURLConnection connection;
36 private final java.util.concurrent.atomic.AtomicReference<Result> result=new java.util.concurrent.atomic.AtomicReference<Result>();
37 private final java.util.concurrent.atomic.AtomicBoolean consumed=new java.util.concurrent.atomic.AtomicBoolean();
38 private final java.util.concurrent.atomic.AtomicBoolean notified=new java.util.concurrent.atomic.AtomicBoolean();
39 private volatile Completion completion;
40 private final java.util.concurrent.CountDownLatch ready=new java.util.concurrent.CountDownLatch(1);
41 private TimerTask expiry;
42 private Call(String id,String body,int timeout,int limit,boolean retainAfterWait) {
43 requestId=id;this.body=body.getBytes(StandardCharsets.UTF_8);responseLimit=limit;
44 long requestDeadline=submitted+TimeUnit.MILLISECONDS.toNanos(timeout);
45 deadline=retainAfterWait?requestDeadline:Math.min(JevDeadline.absolute(),requestDeadline);
46 }
47 private boolean finish(String raw,String error,int status) {
48 if(!result.compareAndSet(null,new Result(this,raw,error,status)))return false;
49 ready.countDown();
50 if(expiry!=null)expiry.cancel();notifyCompletion();return true;
51 }
52 // The listener only enqueues immutable receipt data; it must never wait or touch GUI state.
53 void onComplete(Completion listener) {completion=listener;notifyCompletion();}
54 private void notifyCompletion() {
55 Completion listener=completion;Result value=result.get();
56 if(listener!=null && value!=null && notified.compareAndSet(false,true))
57 listener.completed(value);
58 }
59 public Result poll() {
60 if(result.get()==null && System.nanoTime()>=deadline)cancel("TIMEOUT");
61 return result.get();
62 }
63 /** Return a completed result once. A completed HTTP call is not a stream. */
64 public Result consume() {
65 Result value=poll();
66 return value!=null && consumed.compareAndSet(false,true)?value:null;
67 }
68 public Result await() {
69 if(poll()==null)try {
70 if(!ready.await(Math.max(0,deadline-System.nanoTime()),TimeUnit.NANOSECONDS))cancel("TIMEOUT");
71 } catch(InterruptedException interrupted){cancel("INTERRUPTED");Thread.currentThread().interrupt();}
72 return result.get();
73 }
74 /** End the caller's wait without cancelling an already submitted request. */
75 public Result awaitUntil(long waitDeadline) {
76 if(poll()==null)try {
77 ready.await(Math.max(0,Math.min(waitDeadline,deadline)-System.nanoTime()),TimeUnit.NANOSECONDS);
78 }catch(InterruptedException e){Thread.currentThread().interrupt();}
79 return poll();
80 }
81 public void cancel(String reason) {
82 if(finish(null,reason==null?"CANCELLED":reason,0)) {
83 waiting.remove(this);
84 HttpURLConnection current=connection;if(current!=null)disconnects.offer(current);
85 }
86 }
87 private int remaining() throws SocketTimeoutException {
88 long nanos=deadline-System.nanoTime();
89 if(nanos<=0 || poll()!=null)throw new SocketTimeoutException();
90 return (int)Math.min(Integer.MAX_VALUE,Math.max(1,TimeUnit.NANOSECONDS.toMillis(nanos)));
91 }
92 private void execute() {
93 if(poll()!=null)return;
94 int status=0;
95 try {
96 connection=(HttpURLConnection)endpoint.openConnection();
97 if(poll()!=null)return;sent=System.nanoTime();
98 connection.setConnectTimeout(remaining());connection.setReadTimeout(remaining());
99 connection.setInstanceFollowRedirects(false);connection.setRequestMethod("POST");
100 connection.setRequestProperty("Content-Type","application/json; charset=utf-8");
101 connection.setRequestProperty("X-StartLux-Request-Class",requestClass);
102 connection.setRequestProperty("X-APE-Request-ID",requestId);
103 for(Map.Entry<String,String> field:snapshotIdentity.entrySet())
104 connection.setRequestProperty("X-APE-"+field.getKey().replace('_','-'),field.getValue());
105 connection.setRequestProperty("Accept","application/json");connection.setRequestProperty("Connection","close");
106 if(!token.isEmpty())connection.setRequestProperty("Authorization","Bearer "+token);
107 connection.setDoOutput(true);connection.setFixedLengthStreamingMode(body.length);
108 OutputStream stream=connection.getOutputStream();
109 try{stream.write(body);}finally{stream.close();}
110 connection.setReadTimeout(remaining());status=connection.getResponseCode();
111 if(status!=200){finish(null,"HTTP_"+status,status);return;}
112 ByteArrayOutputStream bytes=new ByteArrayOutputStream();byte[] buffer=new byte[4096];
113 InputStream input=connection.getInputStream();
114 try {
115 while(true) {
116 connection.setReadTimeout(remaining());int size=input.read(buffer);
117 if(size<0)break;
118 if(size>responseLimit-bytes.size()){finish(null,"RESPONSE_TOO_LARGE",status);return;}
119 bytes.write(buffer,0,size);
120 }
121 }finally{input.close();}
122 if(System.nanoTime()>=deadline)cancel("TIMEOUT");
123 else finish(new String(bytes.toByteArray(),StandardCharsets.UTF_8),null,status);
124 }catch(SocketTimeoutException failure){cancel("TIMEOUT");}
125 catch(IOException failure){finish(null,"TRANSPORT_FAILURE",status);}
126 catch(RuntimeException failure){finish(null,"TRANSPORT_FAILURE",status);}
127 finally{HttpURLConnection current=connection;connection=null;if(current!=null)current.disconnect();}
128 }
129 }
130 private final URL endpoint;
131 private final String token,requestClass;
132 private final ArrayBlockingQueue<Call> waiting;
133 private final ArrayBlockingQueue<HttpURLConnection> disconnects=new ArrayBlockingQueue<HttpURLConnection>(1);
134 private final Timer deadlines;
135 private final Thread worker,cleanup;
136 private volatile boolean closed;
137 private volatile Call active;
138 private volatile Map<String,String> snapshotIdentity=Collections.emptyMap();
139 private boolean used,identitySet;
140 private final java.util.concurrent.locks.ReentrantLock submitLock=new java.util.concurrent.locks.ReentrantLock();
141
142 public void setSnapshotIdentity(Map<String,String> identity) {JevDeadline.lock(submitLock);try {
143 if(used || identitySet)throw new IllegalStateException("Snapshot identity must precede all requests");
144 Map<String,String> copy=new LinkedHashMap<String,String>();
145 for(Map.Entry<String,String> field:identity.entrySet()) {
146 if(!field.getKey().matches("experiment|device|app|rep|branch|checkpoint_id")
147 || !field.getValue().matches("[A-Za-z0-9_-]{1,160}"))throw new IllegalArgumentException("Invalid Snapshot HTTP identity");
148 copy.put(field.getKey(),field.getValue());
149 }
150 snapshotIdentity=Collections.unmodifiableMap(copy);identitySet=true;
151 } finally {submitLock.unlock();}}
152
153 public JevTransport(String endpoint,String token,int capacity,String requestClass)throws IOException {
154 if(!"probe".equals(requestClass) && !"region".equals(requestClass))throw new IllegalArgumentException("invalid request class");
155 this.requestClass=requestClass;
156 if(capacity<1 || capacity>64)throw new IllegalArgumentException("invalid request capacity");
157 this.endpoint=new URL(endpoint);
158 String host=this.endpoint.getHost(),scheme=this.endpoint.getProtocol();
159 if(this.endpoint.getUserInfo()!=null || !("https".equals(scheme) || "http".equals(scheme)
160 && ("127.0.0.1".equals(host) || "localhost".equals(host) || "211.71.15.47".equals(host))))
161 throw new IllegalArgumentException("invalid decision endpoint");
162 this.token=token==null?"":token;waiting=new ArrayBlockingQueue<Call>(capacity);
163 deadlines=new Timer("ape-"+requestClass+"-deadline",true);
164 worker=new Thread(new Runnable(){public void run(){
165 while(!closed)try{Call call=waiting.take();active=call;try{if(closed)call.cancel("CLOSED");else call.execute();}finally{active=null;}}
166 catch(InterruptedException interrupted){if(closed)return;}
167 }},"ape-"+requestClass+"-http");worker.setDaemon(true);worker.start();
168 cleanup=new Thread(new Runnable(){public void run(){
169 while(!closed || !disconnects.isEmpty())try {
170 HttpURLConnection connection=disconnects.poll(100,TimeUnit.MILLISECONDS);if(connection!=null)connection.disconnect();
171 }catch(InterruptedException interrupted){if(closed && disconnects.isEmpty())return;}
172 catch(RuntimeException ignored){ }
173 }},"ape-"+requestClass+"-cleanup");cleanup.setDaemon(true);cleanup.start();
174 }
175 public Call submit(String requestId,String body,int timeoutMs,int maxResponseBytes) {
176 return submit(requestId,body,timeoutMs,maxResponseBytes,false);
177 }
178 /** Only the network lifetime outlives the caller's budget; enqueue remains budgeted. */
179 Call submitRetained(String requestId,String body,int timeoutMs,int maxResponseBytes) {
180 return submit(requestId,body,timeoutMs,maxResponseBytes,true);
181 }
182 private Call submit(String requestId,String body,int timeoutMs,int maxResponseBytes,boolean retained) {JevDeadline.lock(submitLock);try {
183 if(requestId==null || requestId.isEmpty() || body==null || timeoutMs<=0 || maxResponseBytes<=0)
184 throw new IllegalArgumentException("invalid request bounds");
185 if(closed)return null;
186 JevDeadline.check();used=true;
187 final Call call=new Call(requestId,body,timeoutMs,maxResponseBytes,retained);
188 call.expiry=new TimerTask(){public void run(){call.cancel("TIMEOUT");}};
189 // Schedule before publishing: an immediate response may cancel this task.
190 deadlines.schedule(call.expiry,Math.max(1,TimeUnit.NANOSECONDS.toMillis(Math.max(0,call.deadline-System.nanoTime()))));
191 if(!waiting.offer(call)){call.cancel("QUEUE_FULL");return null;}
192 return call;
193 } finally {submitLock.unlock();}}
194 int outstanding(){Call call=active;return waiting.size()+(call!=null && call.poll()==null?1:0);}
195 public void close() {JevDeadline.lock(submitLock);try {
196 if(closed)return;closed=true;
197 Call call=active;if(call!=null)call.cancel("CLOSED");
198 while((call=waiting.poll())!=null)call.cancel("CLOSED");
199 deadlines.cancel();worker.interrupt();cleanup.interrupt();
200 } finally {submitLock.unlock();}}
201}