Phase 4：RFC 3261 §17 四个事务 FSM + 时间轮 + 事务表

sip-transaction/pom.xml

@@ -26,9 +26,18 @@
             <groupId>com.sip</groupId>
             <artifactId>sip-transport-api</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.sip</groupId>
+            <artifactId>sip-codec</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-simple</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>

sip-transaction/src/main/java/com/sip/transaction/ClientTransactionListener.java

@@ -0,0 +1,24 @@
+package com.sip.transaction;
+
+import com.sip.message.SipResponse;
+
+/**
+ * Callbacks delivered by a client transaction to the Transaction User (TU).
+ *
+ * <p>The contract is one-shot per event; implementations should hand work
+ * off to a virtual thread if they need to do anything non-trivial.</p>
+ */
+public interface ClientTransactionListener {
+
+    /** Provisional response arrived (1xx). */
+    default void onProvisional(SipResponse response) { }
+
+    /** Final response arrived (2xx-6xx). */
+    void onFinal(SipResponse response);
+
+    /** Transaction timed out (Timer B / F). */
+    void onTimeout();
+
+    /** Transport-level failure observed; transaction is terminated. */
+    default void onTransportError(Throwable error) { }
+}

sip-transaction/src/main/java/com/sip/transaction/InviteClientTransaction.java

@@ -0,0 +1,237 @@
+package com.sip.transaction;
+
+import com.sip.message.SipMethod;
+import com.sip.message.SipRequest;
+import com.sip.message.SipResponse;
+import com.sip.message.SipVersion;
+import com.sip.message.header.HeaderName;
+import com.sip.message.header.Headers;
+import com.sip.transport.Endpoint;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * INVITE client transaction FSM — RFC 3261 §17.1.1.
+ *
+ * <p>Key complications versus non-INVITE:</p>
+ * <ul>
+ *   <li>Timer A doubles on each retransmit (cap at T2 not applied per spec,
+ *       but at 64*T1 implicit via Timer B).</li>
+ *   <li>Non-2xx responses require the TU's ACK to be sent automatically by
+ *       this FSM (§17.1.1.3 "ACK for non-2xx is part of the transaction").</li>
+ *   <li>2xx responses ARE NOT handled by this transaction — they go directly
+ *       to the TU which builds a new ACK on a separate transaction.</li>
+ *   <li>Timer D in the Completed state allows late retransmits of the final
+ *       to be absorbed.</li>
+ * </ul>
+ */
+public final class InviteClientTransaction implements Transaction {
+
+    private static final Logger log = LoggerFactory.getLogger(InviteClientTransaction.class);
+
+    private final TransactionKey key;
+    private final SipRequest request;
+    private final Endpoint peer;
+    private final boolean reliable;
+    private final MessageSender sender;
+    private final TimerScheduler timers;
+    private final ClientTransactionListener listener;
+
+    private final AtomicReference<TransactionState> state =
+            new AtomicReference<>(TransactionState.CALLING);
+    private final AtomicInteger attempt = new AtomicInteger(0);
+
+    private volatile TimerScheduler.Handle timerA;
+    private volatile TimerScheduler.Handle timerB;
+    private volatile TimerScheduler.Handle timerD;
+    private volatile SipResponse finalResponse;
+
+    private InviteClientTransaction(SipRequest request, Endpoint peer,
+                                     MessageSender sender, TimerScheduler timers,
+                                     ClientTransactionListener listener) {
+        this.request = request;
+        this.peer = peer;
+        this.reliable = peer.transport().isReliable();
+        this.sender = sender;
+        this.timers = timers;
+        this.listener = listener;
+        this.key = TransactionKey.of(request);
+    }
+
+    public static InviteClientTransaction start(SipRequest request, Endpoint peer,
+                                                 MessageSender sender,
+                                                 TimerScheduler timers,
+                                                 ClientTransactionListener listener) {
+        if (request.method() != SipMethod.INVITE) {
+            throw new IllegalArgumentException(
+                    "InviteClientTransaction requires INVITE, got " + request.method());
+        }
+        InviteClientTransaction tx = new InviteClientTransaction(
+                request, peer, sender, timers, listener);
+        tx.fire();
+        return tx;
+    }
+
+    @Override
+    public TransactionKey key() { return key; }
+
+    @Override
+    public TransactionState state() { return state.get(); }
+
+    private void fire() {
+        sender.send(request, peer).whenComplete((ok, err) -> {
+            if (err != null) {
+                terminateWith(() -> listener.onTransportError(err));
+            }
+        });
+        if (!reliable) {
+            scheduleA();
+        }
+        timerB = timers.schedule(this::onTimerB, Timing.TIMER_B_MS);
+    }
+
+    private void scheduleA() {
+        long delay = Timing.timerA(attempt.getAndIncrement());
+        timerA = timers.schedule(this::onTimerA, delay);
+    }
+
+    private void onTimerA() {
+        if (state.get() != TransactionState.CALLING) {
+            return;
+        }
+        log.debug("INVITE client retransmit (Timer A) {} attempt={}", key, attempt.get());
+        sender.send(request, peer);
+        scheduleA();
+    }
+
+    private void onTimerB() {
+        if (state.get() == TransactionState.TERMINATED
+                || state.get() == TransactionState.COMPLETED) {
+            return;
+        }
+        log.debug("INVITE client timeout (Timer B) {}", key);
+        terminateWith(listener::onTimeout);
+    }
+
+    private void onTimerD() {
+        terminateWith(() -> { });
+    }
+
+    /** Feed an inbound response into the FSM. */
+    public void onResponse(SipResponse response) {
+        TransactionState s = state.get();
+        if (s == TransactionState.TERMINATED) {
+            return;
+        }
+        int status = response.status();
+
+        if (status < 200) {
+            if (state.compareAndSet(TransactionState.CALLING, TransactionState.PROCEEDING)) {
+                cancelTimer(timerA);
+                listener.onProvisional(response);
+            } else if (s == TransactionState.PROCEEDING) {
+                listener.onProvisional(response);
+            }
+            return;
+        }
+
+        // Final response.
+        if (status >= 200 && status < 300) {
+            // 2xx is handled by the TU — this transaction terminates immediately.
+            if (state.compareAndSet(TransactionState.CALLING, TransactionState.TERMINATED)
+                    || state.compareAndSet(TransactionState.PROCEEDING,
+                            TransactionState.TERMINATED)) {
+                cancelAllTimers();
+                listener.onFinal(response);
+            }
+            return;
+        }
+
+        // 3xx-6xx
+        if (state.compareAndSet(TransactionState.CALLING, TransactionState.COMPLETED)
+                || state.compareAndSet(TransactionState.PROCEEDING,
+                        TransactionState.COMPLETED)) {
+            cancelTimer(timerA);
+            this.finalResponse = response;
+            sendAckForNonSuccess(response);
+            listener.onFinal(response);
+            long d = reliable ? Timing.TIMER_D_MS_RELIABLE : Timing.TIMER_D_MS_UDP;
+            if (d == 0) {
+                terminateWith(() -> { });
+            } else {
+                timerD = timers.schedule(this::onTimerD, d);
+            }
+        } else if (state.get() == TransactionState.COMPLETED) {
+            // Retransmitted final — resend ACK.
+            SipResponse last = this.finalResponse;
+            if (last != null) {
+                sendAckForNonSuccess(last);
+            }
+        }
+    }
+
+    /**
+     * RFC 3261 §17.1.1.3 — ACK for non-2xx is built from the INVITE's
+     * From, Call-ID, CSeq (with method=ACK), top Via, and the response's
+     * To header. The ACK is sent within this transaction on the same
+     * transport / branch.
+     */
+    private void sendAckForNonSuccess(SipResponse response) {
+        Headers original = request.headers();
+        Headers.Builder b = Headers.builder();
+        // Top Via must be reused so the ACK matches our outbound branch.
+        original.first(HeaderName.VIA).ifPresent(via -> b.add(via));
+        original.first(HeaderName.FROM).ifPresent(from -> b.add(from));
+        response.headers().first(HeaderName.TO).ifPresentOrElse(
+                b::add,
+                () -> original.first(HeaderName.TO).ifPresent(b::add));
+        original.first(HeaderName.CALL_ID).ifPresent(b::add);
+        original.first(HeaderName.CSEQ).ifPresent(cseq -> {
+            // Replace CSeq method with ACK.
+            String value = cseq.value();
+            int sp = -1;
+            for (int i = 0; i < value.length(); i++) {
+                if (Character.isWhitespace(value.charAt(i))) {
+                    sp = i;
+                    break;
+                }
+            }
+            String seq = sp < 0 ? value : value.substring(0, sp);
+            b.add(HeaderName.CSEQ, seq + " ACK");
+        });
+        original.first(HeaderName.MAX_FORWARDS).ifPresent(b::add);
+
+        SipRequest ack = new SipRequest(
+                SipMethod.ACK,
+                request.requestUri(),
+                SipVersion.SIP_2_0,
+                b.build(),
+                new byte[0]);
+        sender.send(ack, peer);
+    }
+
+    private void terminateWith(Runnable hook) {
+        if (state.getAndSet(TransactionState.TERMINATED) == TransactionState.TERMINATED) {
+            return;
+        }
+        cancelAllTimers();
+        try {
+            hook.run();
+        } catch (Throwable t) {
+            log.warn("listener callback threw on terminate: {}", t.toString());
+        }
+    }
+
+    private void cancelAllTimers() {
+        cancelTimer(timerA);
+        cancelTimer(timerB);
+        cancelTimer(timerD);
+    }
+
+    private static void cancelTimer(TimerScheduler.Handle h) {
+        if (h != null) h.cancel();
+    }
+}