forked from
lass.pds.lassul.us/superconfig
This repository has no description
33 kB
908 lines
1package us.lassul.assist;
2
3import android.content.Context;
4import android.os.Handler;
5import android.os.Looper;
6import android.os.SystemClock;
7import android.util.Log;
8import java.io.BufferedReader;
9import java.io.InputStreamReader;
10import java.net.HttpURLConnection;
11import java.nio.ByteBuffer;
12import java.nio.ByteOrder;
13import java.nio.ShortBuffer;
14import java.nio.charset.StandardCharsets;
15import java.util.ArrayList;
16import java.util.Collections;
17import java.util.List;
18import java.util.concurrent.ExecutorService;
19import java.util.concurrent.Executors;
20import java.util.regex.Pattern;
21import org.json.JSONArray;
22import org.json.JSONException;
23import org.json.JSONObject;
24import org.webrtc.AudioSource;
25import org.webrtc.AudioTrack;
26import org.webrtc.AudioTrackSink;
27import org.webrtc.DataChannel;
28import org.webrtc.IceCandidate;
29import org.webrtc.MediaConstraints;
30import org.webrtc.MediaStream;
31import org.webrtc.MediaStreamTrack;
32import org.webrtc.PeerConnection;
33import org.webrtc.PeerConnectionFactory;
34import org.webrtc.RtpReceiver;
35import org.webrtc.SdpObserver;
36import org.webrtc.SessionDescription;
37import org.webrtc.audio.JavaAudioDeviceModule;
38
39/**
40 * One call with gpt-live-1 over WebRTC, delegating to Hermes through hermes-voice: the web page's
41 * Call (static/app.js) in Java, with the same events, delegation, voice approvals, end_call and
42 * idle rules. Lives on the main thread; WebRTC and network callbacks are posted to it.
43 */
44final class VoiceCall {
45 interface Listener {
46 /** The voice model listens (session.started). */
47 void onListening();
48
49 /** The call is over; the reason is for the log. */
50 void onFinished(String reason, boolean failed);
51 }
52
53 // The voice model accepts at most 500 tokens per append; ~4 chars per token.
54 private static final int APPEND_CHAR_LIMIT = 1400;
55 // Transcript deltas can trail the delegation event; wait for the last words.
56 private static final long DELEGATION_SETTLE_MS = 400;
57 // Once Hermes ended the call (end_call item), the call closes when the voice has been quiet
58 // this long after speaking the reply, at the latest after END_CALL_MAX_MS.
59 private static final long END_CALL_QUIET_MS = 800;
60 private static final long END_CALL_MAX_MS = 30_000;
61 private static final long CLOSE_TIMEOUT_MS = 15_000;
62 // Trickle is fine: the vendor answers with the candidates it has.
63 private static final long ICE_TIMEOUT_MS = 3_000;
64 private static final long TICK_MS = 100;
65 // The voice counts as speaking while a sample above this came within SPEAKING_HOLD_MS.
66 private static final int SPEAKING_PEAK = 1500;
67 private static final long SPEAKING_HOLD_MS = 400;
68 // Spoken answer to an approval question. Denials win, so "no, don't do it" never approves;
69 // anything unclear is asked again.
70 private static final Pattern DENY = Pattern.compile(
71 "\\b(no|nope|deny|denied|don'?t|do not|stop|cancel|abort|reject|nein|nicht)\\b",
72 Pattern.CASE_INSENSITIVE);
73 private static final Pattern APPROVE = Pattern.compile(
74 "\\b(yes|yeah|yep|approve|approved|go ahead|do it|okay|ok|sure|confirm|allow|ja)\\b",
75 Pattern.CASE_INSENSITIVE);
76
77 private static boolean webrtcLoaded;
78
79 private final Context context;
80 private final Api api;
81 private final Conversation conversation;
82 private final Listener listener;
83 private final Handler main = new Handler(Looper.getMainLooper());
84 private final ExecutorService io = Executors.newCachedThreadPool();
85
86 private JavaAudioDeviceModule audio;
87 private PeerConnectionFactory factory;
88 private PeerConnection pc;
89 private AudioSource source;
90 private AudioTrack mic;
91 private DataChannel events;
92 private boolean offerSent;
93 private long idleMs = 90_000;
94
95 // Written by WebRTC's audio thread, read by tick().
96 private volatile long lastLoud;
97 private long lastActivity = SystemClock.elapsedRealtime();
98 private boolean speaking;
99 private String lastSpeaker;
100 private Delegation delegation;
101 private Approval approval;
102 private boolean closed;
103 private boolean closing;
104 private boolean ending;
105 private boolean endingSpoken;
106 private int eventCounter;
107 private double usageSeconds;
108
109 private final Runnable sendOffer = new Runnable() {
110 @Override
111 public void run() {
112 sendOffer();
113 }
114 };
115 private final Runnable tick = new Runnable() {
116 @Override
117 public void run() {
118 tick();
119 }
120 };
121 private final Runnable closeTimedOut = new Runnable() {
122 @Override
123 public void run() {
124 finish("close timed out", false);
125 }
126 };
127 private final Runnable endNow = new Runnable() {
128 @Override
129 public void run() {
130 close();
131 }
132 };
133
134 /** A Hermes turn: the /delegate stream for one delegation of the voice model. */
135 private static final class Delegation {
136 final String id;
137 final StringBuilder answer = new StringBuilder();
138 String failure;
139 boolean endCall;
140 volatile boolean aborted;
141 volatile HttpURLConnection connection;
142
143 Delegation(String id) {
144 this.id = id;
145 }
146 }
147
148 private static final class Approval {
149 final String delegationId;
150 final String code;
151
152 Approval(String delegationId, String code) {
153 this.delegationId = delegationId;
154 this.code = code;
155 }
156 }
157
158 VoiceCall(Context context, Api api, Conversation conversation, Listener listener) {
159 this.context = context;
160 this.api = api;
161 this.conversation = conversation;
162 this.listener = listener;
163 }
164
165 void start() {
166 io.execute(new Runnable() {
167 @Override
168 public void run() {
169 try {
170 final long idle = api.get("/config").optLong("idle_seconds", 90) * 1000;
171 main.post(new Runnable() {
172 @Override
173 public void run() {
174 idleMs = idle;
175 connect();
176 }
177 });
178 } catch (Exception error) {
179 failLater("Settings not loaded: " + error.getMessage());
180 }
181 }
182 });
183 }
184
185 private void connect() {
186 if (closed) {
187 return;
188 }
189 if (!webrtcLoaded) {
190 PeerConnectionFactory.initialize(PeerConnectionFactory.InitializationOptions
191 .builder(context.getApplicationContext())
192 .createInitializationOptions());
193 webrtcLoaded = true;
194 }
195 audio = JavaAudioDeviceModule.builder(context)
196 .setUseHardwareAcousticEchoCanceler(true)
197 .setUseHardwareNoiseSuppressor(true)
198 .createAudioDeviceModule();
199 factory = PeerConnectionFactory.builder()
200 .setAudioDeviceModule(audio)
201 .createPeerConnectionFactory();
202 PeerConnection.RTCConfiguration config =
203 new PeerConnection.RTCConfiguration(new ArrayList<PeerConnection.IceServer>());
204 config.sdpSemantics = PeerConnection.SdpSemantics.UNIFIED_PLAN;
205 pc = factory.createPeerConnection(config, new PeerObserver());
206 if (pc == null) {
207 fail("WebRTC refused the connection");
208 return;
209 }
210 source = factory.createAudioSource(new MediaConstraints());
211 mic = factory.createAudioTrack("mic", source);
212 pc.addTrack(mic, Collections.singletonList("mic"));
213 // The data channel must exist before the offer so its m-line is negotiated.
214 events = pc.createDataChannel("oai-events", new DataChannel.Init());
215 events.registerObserver(new EventsObserver());
216 pc.createOffer(new Sdp("offer") {
217 @Override
218 public void onCreateSuccess(final SessionDescription offer) {
219 main.post(new Runnable() {
220 @Override
221 public void run() {
222 if (closed) {
223 return;
224 }
225 pc.setLocalDescription(new Sdp("local description") {
226 @Override
227 public void onSetSuccess() {
228 main.postDelayed(sendOffer, ICE_TIMEOUT_MS);
229 }
230 }, offer);
231 }
232 });
233 }
234 }, new MediaConstraints());
235 }
236
237 private void sendOffer() {
238 if (offerSent || closed) {
239 return;
240 }
241 offerSent = true;
242 main.removeCallbacks(sendOffer);
243 final JSONObject body;
244 try {
245 body = new JSONObject()
246 .put("sdp", pc.getLocalDescription().description)
247 .put("history", Conversation.toJson(conversation.lines));
248 } catch (JSONException error) {
249 fail(error.toString());
250 return;
251 }
252 io.execute(new Runnable() {
253 @Override
254 public void run() {
255 try {
256 final String answer = api.post("/session", body).getString("sdp");
257 main.post(new Runnable() {
258 @Override
259 public void run() {
260 if (closed) {
261 return;
262 }
263 pc.setRemoteDescription(new Sdp("remote description"),
264 new SessionDescription(SessionDescription.Type.ANSWER, answer));
265 main.postDelayed(tick, TICK_MS);
266 }
267 });
268 } catch (Exception error) {
269 failLater(error.getMessage());
270 }
271 }
272 });
273 }
274
275 private void tick() {
276 if (closed) {
277 return;
278 }
279 long now = SystemClock.elapsedRealtime();
280 long quiet = now - lastLoud;
281 speaking = quiet < SPEAKING_HOLD_MS;
282 if (speaking) {
283 lastActivity = now;
284 }
285 if (ending && endingSpoken && quiet >= END_CALL_QUIET_MS) {
286 close();
287 } else if (delegation == null && !speaking && now - lastActivity > idleMs) {
288 Log.i(CallService.TAG, "hanging up (idle)");
289 close();
290 }
291 main.postDelayed(tick, TICK_MS);
292 }
293
294 private boolean send(JSONObject event) {
295 if (events == null || events.state() != DataChannel.State.OPEN) {
296 return false;
297 }
298 try {
299 eventCounter += 1;
300 event.put("event_id", "evt_" + eventCounter);
301 } catch (JSONException error) {
302 return false;
303 }
304 byte[] bytes = event.toString().getBytes(StandardCharsets.UTF_8);
305 return events.send(new DataChannel.Buffer(ByteBuffer.wrap(bytes), false));
306 }
307
308 private void append(String type, String delegationId, String content) {
309 try {
310 send(new JSONObject()
311 .put("type", type)
312 .put("delegation_id", delegationId)
313 .put("content", truncate(content)));
314 } catch (JSONException error) {
315 Log.w(CallService.TAG, "append failed", error);
316 }
317 }
318
319 private void onEvent(JSONObject event) {
320 switch (event.optString("type")) {
321 case "session.started":
322 listener.onListening();
323 break;
324 case "session.input_transcript.delta":
325 lastActivity = SystemClock.elapsedRealtime();
326 addTranscript("user", event.optString("delta"));
327 break;
328 case "session.output_transcript.delta":
329 lastActivity = SystemClock.elapsedRealtime();
330 addTranscript("assistant", event.optString("delta"));
331 if (ending) {
332 endingSpoken = true;
333 }
334 break;
335 case "session.delegation.created":
336 JSONObject created = event.optJSONObject("delegation");
337 if (created != null) {
338 delegate(created.optString("id"));
339 }
340 break;
341 case "session.usage.updated":
342 usageSeconds = usageOf(event, usageSeconds);
343 break;
344 case "error":
345 JSONObject error = event.optJSONObject("error");
346 conversation.add("error", error != null
347 ? error.optString("message", "voice model error")
348 : "voice model error");
349 break;
350 case "session.closed":
351 usageSeconds = usageOf(event, usageSeconds);
352 finish(event.optString("reason", "closed"), false);
353 break;
354 default:
355 break;
356 }
357 }
358
359 private static double usageOf(JSONObject event, double fallback) {
360 JSONObject usage = event.optJSONObject("usage");
361 return usage != null ? usage.optDouble("seconds", fallback) : fallback;
362 }
363
364 // Deltas are appended exactly as received; a speaker change starts a new line.
365 private void addTranscript(String speaker, String delta) {
366 List<Conversation.Line> lines = conversation.lines;
367 Conversation.Line last = lines.isEmpty() ? null : lines.get(lines.size() - 1);
368 if (last != null && last.speaker.equals(speaker) && speaker.equals(lastSpeaker)) {
369 last.text += delta;
370 conversation.save();
371 } else {
372 conversation.add(speaker, delta);
373 }
374 lastSpeaker = speaker;
375 }
376
377 private void delegate(String id) {
378 // One Hermes turn at a time: a newer delegation (usually a correction) supersedes the
379 // running one, and the server stops the old run. While Hermes waits for an approval,
380 // the next delegation is the user's answer, not a new request.
381 if (approval != null) {
382 answerApproval(id);
383 return;
384 }
385 final boolean replace = delegation != null;
386 if (delegation != null) {
387 abort(delegation);
388 }
389 final Delegation current = new Delegation(id);
390 delegation = current;
391 main.postDelayed(new Runnable() {
392 @Override
393 public void run() {
394 runDelegation(current, replace);
395 }
396 }, DELEGATION_SETTLE_MS);
397 }
398
399 private void runDelegation(final Delegation current, boolean replace) {
400 // A hang-up meanwhile sent these lines along itself.
401 if (delegation != current || closed) {
402 return;
403 }
404 List<Conversation.Line> lines = conversation.undelegated("user", "assistant");
405 boolean userSpoke = false;
406 for (Conversation.Line line : lines) {
407 userSpoke |= line.speaker.equals("user");
408 }
409 if (!userSpoke) {
410 List<Conversation.Line> dialogue = new ArrayList<>();
411 for (Conversation.Line line : conversation.lines) {
412 if (line.speaker.equals("user") || line.speaker.equals("assistant")) {
413 dialogue.add(line);
414 }
415 }
416 lines = dialogue.subList(Math.max(0, dialogue.size() - 4), dialogue.size());
417 }
418 conversation.delegated = conversation.lines.size();
419 conversation.save();
420 final JSONObject body;
421 try {
422 body = new JSONObject()
423 .put("hermes_session", conversation.hermesSession)
424 .put("lines", Conversation.toJson(lines))
425 .put("replace", replace);
426 } catch (JSONException error) {
427 return;
428 }
429 io.execute(new Runnable() {
430 @Override
431 public void run() {
432 stream(current, body);
433 }
434 });
435 }
436
437 /** Network thread: reads /delegate's NDJSON and posts each item to the main thread. */
438 private void stream(final Delegation current, JSONObject body) {
439 String failure = null;
440 HttpURLConnection connection = null;
441 try {
442 connection = api.stream("/delegate", body);
443 current.connection = connection;
444 if (current.aborted) {
445 return;
446 }
447 BufferedReader reader = new BufferedReader(
448 new InputStreamReader(connection.getInputStream(), StandardCharsets.UTF_8));
449 for (String line; (line = reader.readLine()) != null; ) {
450 // Blank lines are the server's keepalives.
451 if (line.trim().isEmpty()) {
452 continue;
453 }
454 final JSONObject item = new JSONObject(line);
455 main.post(new Runnable() {
456 @Override
457 public void run() {
458 onItem(current, item);
459 }
460 });
461 }
462 } catch (Exception error) {
463 if (current.aborted) {
464 return;
465 }
466 failure = String.valueOf(error.getMessage());
467 } finally {
468 if (connection != null) {
469 connection.disconnect();
470 }
471 }
472 final String streamFailure = failure;
473 main.post(new Runnable() {
474 @Override
475 public void run() {
476 onDone(current, streamFailure);
477 }
478 });
479 }
480
481 private void onItem(Delegation current, JSONObject item) {
482 if (delegation != current) {
483 return;
484 }
485 lastActivity = SystemClock.elapsedRealtime();
486 String type = item.optString("type");
487 if (type.equals("approval")) {
488 askApproval(current.id, item);
489 return;
490 }
491 // Anything after an approval request means Hermes moved on (answered here, from Matrix,
492 // or by its own timeout).
493 approval = null;
494 if (type.equals("tool") && item.optString("status").equals("running")) {
495 append("session.thinking.append", current.id,
496 "Hermes is working on it: " + item.optString("label"));
497 } else if (type.equals("end_call")) {
498 current.endCall = true;
499 } else if (type.equals("delta")) {
500 current.answer.append(item.optString("text"));
501 } else if (type.equals("done") && item.has("error") && !item.isNull("error")) {
502 current.failure = item.optString("error");
503 } else if (type.equals("error")) {
504 current.failure = item.optString("message");
505 }
506 }
507
508 private void onDone(Delegation current, String streamFailure) {
509 if (delegation != current) {
510 return;
511 }
512 delegation = null;
513 approval = null;
514 lastActivity = SystemClock.elapsedRealtime();
515 String failure = current.failure != null ? current.failure : streamFailure;
516 String answer = current.answer.toString().trim();
517 if (failure != null && answer.isEmpty()) {
518 conversation.add("error", failure);
519 append("session.commentary.append", current.id,
520 "The request to Hermes failed: " + failure);
521 } else {
522 String text = answer.isEmpty() ? "Hermes finished without an answer." : answer;
523 conversation.add("hermes", text);
524 for (String chunk : chunks(text)) {
525 append("session.commentary.append", current.id, chunk);
526 }
527 }
528 if (current.endCall) {
529 // Hermes ended the call: close it once the voice has spoken the reply just handed over.
530 ending = true;
531 endingSpoken = false;
532 main.postDelayed(endNow, END_CALL_MAX_MS);
533 }
534 }
535
536 private void abort(final Delegation current) {
537 current.aborted = true;
538 io.execute(new Runnable() {
539 @Override
540 public void run() {
541 HttpURLConnection connection = current.connection;
542 if (connection != null) {
543 connection.disconnect();
544 }
545 }
546 });
547 }
548
549 private void askApproval(String id, JSONObject item) {
550 approval = new Approval(id, item.isNull("code") ? null : item.optString("code", null));
551 String what = item.optString("description");
552 if (what.isEmpty()) {
553 what = "run a command";
554 }
555 String command = item.isNull("command") ? "" : item.optString("command");
556 conversation.add("hermes",
557 "Approval needed: " + what + (command.isEmpty() ? "" : "\n" + command));
558 append("session.commentary.append", id,
559 "Hermes needs the user's approval before it continues: " + what + "."
560 + (command.isEmpty() ? "" : " The command is: " + command + ".")
561 + " Ask the user whether to approve or deny it.");
562 }
563
564 private void answerApproval(final String id) {
565 main.postDelayed(new Runnable() {
566 @Override
567 public void run() {
568 if (closed) {
569 return;
570 }
571 StringBuilder said = new StringBuilder();
572 for (Conversation.Line line : conversation.undelegated("user")) {
573 said.append(line.text).append(' ');
574 }
575 conversation.delegated = conversation.lines.size();
576 conversation.save();
577 String choice = classifyApproval(said.toString());
578 if (choice == null) {
579 append("session.commentary.append", id,
580 "That was not a clear answer. Ask the user again: approve or deny?");
581 return;
582 }
583 respondApproval(choice, id);
584 }
585 }, DELEGATION_SETTLE_MS);
586 }
587
588 static String classifyApproval(String said) {
589 if (DENY.matcher(said).find()) {
590 return "deny";
591 }
592 if (APPROVE.matcher(said).find()) {
593 return "once";
594 }
595 return null;
596 }
597
598 private void respondApproval(final String choice, final String replyId) {
599 final Approval answered = approval;
600 if (answered == null) {
601 return;
602 }
603 approval = null;
604 io.execute(new Runnable() {
605 @Override
606 public void run() {
607 String message;
608 boolean ok = false;
609 try {
610 api.post("/approve", new JSONObject()
611 .put("code", answered.code == null ? JSONObject.NULL : answered.code)
612 .put("choice", choice));
613 ok = true;
614 message = choice.equals("deny")
615 ? "Denied. Hermes will not run it."
616 : "Approved. Hermes is continuing.";
617 } catch (Exception error) {
618 message = "Could not send the answer to Hermes: " + error.getMessage();
619 }
620 final String reply = message;
621 final boolean sent = ok;
622 main.post(new Runnable() {
623 @Override
624 public void run() {
625 conversation.add(sent ? "hermes" : "error", reply);
626 // A fast turn can finish before this reply returns; its result was already
627 // handed to the voice, so an "is continuing" would be stale.
628 if (closed || (sent && delegation == null)) {
629 return;
630 }
631 append("session.commentary.append",
632 replyId != null ? replyId : answered.delegationId, reply);
633 }
634 });
635 }
636 });
637 }
638
639 /** Ends the call gracefully: the voice model closes the session and reports final usage. */
640 void close() {
641 if (closed || closing) {
642 return;
643 }
644 closing = true;
645 try {
646 if (!send(new JSONObject().put("type", "session.close"))) {
647 finish("closed", false);
648 return;
649 }
650 } catch (JSONException error) {
651 finish("closed", false);
652 return;
653 }
654 main.postDelayed(closeTimedOut, CLOSE_TIMEOUT_MS);
655 }
656
657 private void failLater(final String message) {
658 main.post(new Runnable() {
659 @Override
660 public void run() {
661 fail(message);
662 }
663 });
664 }
665
666 private void fail(String message) {
667 if (closed) {
668 return;
669 }
670 Log.w(CallService.TAG, "call failed: " + message);
671 conversation.add("error", String.valueOf(message));
672 finish("failed to start", true);
673 }
674
675 private void finish(String reason, boolean failed) {
676 if (closed) {
677 return;
678 }
679 closed = true;
680 main.removeCallbacksAndMessages(null);
681 hangUp();
682 // Only the stream is dropped; the server keeps following the Hermes run.
683 if (delegation != null) {
684 abort(delegation);
685 delegation = null;
686 }
687 if (events != null) {
688 events.unregisterObserver();
689 events.close();
690 events.dispose();
691 }
692 if (pc != null) {
693 pc.close();
694 pc.dispose();
695 }
696 if (mic != null) {
697 mic.dispose();
698 }
699 if (source != null) {
700 source.dispose();
701 }
702 if (factory != null) {
703 factory.dispose();
704 }
705 if (audio != null) {
706 audio.release();
707 }
708 conversation.billedSeconds += usageSeconds;
709 conversation.save();
710 io.shutdown();
711 Log.i(CallService.TAG, "call ended: " + reason);
712 listener.onFinished(reason, failed);
713 }
714
715 // Tell the server the call is over, with what the user said since the last delegation: a
716 // request spoken right before hanging up still reaches Hermes.
717 private void hangUp() {
718 List<Conversation.Line> lines = conversation.undelegated("user", "assistant");
719 conversation.delegated = conversation.lines.size();
720 conversation.save();
721 final JSONObject body;
722 try {
723 body = new JSONObject()
724 .put("hermes_session", conversation.hermesSession)
725 .put("lines", Conversation.toJson(lines));
726 } catch (JSONException error) {
727 return;
728 }
729 io.execute(new Runnable() {
730 @Override
731 public void run() {
732 try {
733 api.post("/hangup", body);
734 } catch (Exception error) {
735 Log.w(CallService.TAG, "hang-up not delivered: " + error.getMessage());
736 }
737 }
738 });
739 }
740
741 private static String truncate(String text) {
742 return text.length() > APPEND_CHAR_LIMIT ? text.substring(0, APPEND_CHAR_LIMIT) : text;
743 }
744
745 /** Sentences packed into appends of at most APPEND_CHAR_LIMIT characters. */
746 static List<String> chunks(String text) {
747 String clean = text.replaceAll("\\s+", " ").trim();
748 List<String> out = new ArrayList<>();
749 String current = "";
750 for (String sentence : clean.split("(?<=[.!?])\\s+")) {
751 if (sentence.length() > APPEND_CHAR_LIMIT) {
752 if (!current.isEmpty()) {
753 out.add(current);
754 }
755 current = "";
756 for (int i = 0; i < sentence.length(); i += APPEND_CHAR_LIMIT) {
757 out.add(sentence.substring(i, Math.min(sentence.length(), i + APPEND_CHAR_LIMIT)));
758 }
759 } else if (!current.isEmpty()
760 && current.length() + 1 + sentence.length() > APPEND_CHAR_LIMIT) {
761 out.add(current);
762 current = sentence;
763 } else {
764 current = current.isEmpty() ? sentence : current + " " + sentence;
765 }
766 }
767 if (!current.isEmpty()) {
768 out.add(current);
769 }
770 return out;
771 }
772
773 /** Logs failures; a failed description ends the call. */
774 private class Sdp implements SdpObserver {
775 private final String what;
776
777 Sdp(String what) {
778 this.what = what;
779 }
780
781 @Override
782 public void onCreateSuccess(SessionDescription description) {}
783
784 @Override
785 public void onSetSuccess() {}
786
787 @Override
788 public void onCreateFailure(String error) {
789 failLater(what + ": " + error);
790 }
791
792 @Override
793 public void onSetFailure(String error) {
794 failLater(what + ": " + error);
795 }
796 }
797
798 private class PeerObserver implements PeerConnection.Observer {
799 @Override
800 public void onConnectionChange(PeerConnection.PeerConnectionState state) {
801 if (state == PeerConnection.PeerConnectionState.FAILED
802 || state == PeerConnection.PeerConnectionState.DISCONNECTED) {
803 main.post(new Runnable() {
804 @Override
805 public void run() {
806 finish("connection lost", false);
807 }
808 });
809 }
810 }
811
812 @Override
813 public void onIceGatheringChange(PeerConnection.IceGatheringState state) {
814 if (state == PeerConnection.IceGatheringState.COMPLETE) {
815 main.post(sendOffer);
816 }
817 }
818
819 // The voice's audio plays by itself; the sink only measures it, for speaking().
820 @Override
821 public void onAddTrack(RtpReceiver receiver, MediaStream[] streams) {
822 MediaStreamTrack track = receiver.track();
823 if (track instanceof AudioTrack) {
824 ((AudioTrack) track).addSink(new AudioTrackSink() {
825 @Override
826 public void onData(ByteBuffer data, int bitsPerSample, int sampleRate,
827 int channels, int frames, long timestamp) {
828 if (bitsPerSample != 16) {
829 return;
830 }
831 ShortBuffer samples = data.duplicate().order(ByteOrder.nativeOrder())
832 .asShortBuffer();
833 while (samples.hasRemaining()) {
834 if (Math.abs(samples.get()) > SPEAKING_PEAK) {
835 lastLoud = SystemClock.elapsedRealtime();
836 return;
837 }
838 }
839 }
840 });
841 }
842 }
843
844 @Override
845 public void onSignalingChange(PeerConnection.SignalingState state) {}
846
847 @Override
848 public void onIceConnectionChange(PeerConnection.IceConnectionState state) {}
849
850 @Override
851 public void onIceConnectionReceivingChange(boolean receiving) {}
852
853 @Override
854 public void onIceCandidate(IceCandidate candidate) {}
855
856 @Override
857 public void onIceCandidatesRemoved(IceCandidate[] candidates) {}
858
859 @Override
860 public void onAddStream(MediaStream stream) {}
861
862 @Override
863 public void onRemoveStream(MediaStream stream) {}
864
865 @Override
866 public void onDataChannel(DataChannel channel) {}
867
868 @Override
869 public void onRenegotiationNeeded() {}
870 }
871
872 private class EventsObserver implements DataChannel.Observer {
873 @Override
874 public void onMessage(DataChannel.Buffer buffer) {
875 byte[] bytes = new byte[buffer.data.remaining()];
876 buffer.data.get(bytes);
877 final String text = new String(bytes, StandardCharsets.UTF_8);
878 main.post(new Runnable() {
879 @Override
880 public void run() {
881 if (closed) {
882 return;
883 }
884 try {
885 onEvent(new JSONObject(text));
886 } catch (JSONException error) {
887 Log.w(CallService.TAG, "unreadable event", error);
888 }
889 }
890 });
891 }
892
893 @Override
894 public void onStateChange() {
895 if (events.state() == DataChannel.State.CLOSED) {
896 main.post(new Runnable() {
897 @Override
898 public void run() {
899 finish("connection closed", false);
900 }
901 });
902 }
903 }
904
905 @Override
906 public void onBufferedAmountChange(long previous) {}
907 }
908}