SIP Server 第三版实战

nexus.io.sip.rtp
├── codec
│   ├── AudioCodec.java
│   ├── PcmuCodec.java
│   └── PcmaCodec.java
│
├── media
│   ├── AudioFrame.java
│   ├── MediaProcessor.java
│   └── EchoMediaProcessor.java
│
├── packet
│   ├── RtpPacket.java
│   ├── RtpPacketParser.java
│   └── RtpPacketWriter.java
│
├── server
│   ├── RtpUdpHandler.java
│   └── RtpUdpServer.java

package nexus.io.sip.model;

import nexus.io.sip.rtp.RtpUdpServer;
import nexus.io.sip.sdp.CodecSpec;

public class CallSession {

  private String callId;
  private String fromTag;
  private String toTag;

  private String transport;

  private String remoteSipIp;
  private int remoteSipPort;

  private String remoteRtpIp;
  private int remoteRtpPort;
  private int localRtpPort;

  private long createdTime;
  private long updatedTime;
  private long ackDeadline;

  private boolean ackReceived;
  private boolean terminated;

  private String last200Ok;
  private RtpUdpServer rtpServer;

  private CodecSpec selectedCodec;
  private boolean telephoneEventSupported;
  private int remoteTelephoneEventPayloadType = -1;
  private int ptime = 20;

  // 第三阶段新增
  private long localSsrc = System.nanoTime() & 0xFFFFFFFFL;
  private int sendSequence = 0;
  private long sendTimestamp = 0;
  private boolean rtpInitialized = false;

  public synchronized int nextSendSequence() {
    sendSequence = (sendSequence + 1) & 0xFFFF;
    return sendSequence;
  }

  public synchronized long nextSendTimestamp(int sampleCount) {
    if (!rtpInitialized) {
      rtpInitialized = true;
      sendTimestamp = sampleCount;
      return sendTimestamp;
    }
    sendTimestamp = (sendTimestamp + sampleCount) & 0xFFFFFFFFL;
    return sendTimestamp;
  }

  public long getLocalSsrc() {
    return localSsrc;
  }

  public void setLocalSsrc(long localSsrc) {
    this.localSsrc = localSsrc;
  }

  public int getSendSequence() {
    return sendSequence;
  }

  public void setSendSequence(int sendSequence) {
    this.sendSequence = sendSequence;
  }

  public long getSendTimestamp() {
    return sendTimestamp;
  }

  public void setSendTimestamp(long sendTimestamp) {
    this.sendTimestamp = sendTimestamp;
  }

  public boolean isRtpInitialized() {
    return rtpInitialized;
  }

  public void setRtpInitialized(boolean rtpInitialized) {
    this.rtpInitialized = rtpInitialized;
  }

  public String getCallId() {
    return callId;
  }

  public void setCallId(String callId) {
    this.callId = callId;
  }

  public String getFromTag() {
    return fromTag;
  }

  public void setFromTag(String fromTag) {
    this.fromTag = fromTag;
  }

  public String getToTag() {
    return toTag;
  }

  public void setToTag(String toTag) {
    this.toTag = toTag;
  }

  public String getTransport() {
    return transport;
  }

  public void setTransport(String transport) {
    this.transport = transport;
  }

  public String getRemoteSipIp() {
    return remoteSipIp;
  }

  public void setRemoteSipIp(String remoteSipIp) {
    this.remoteSipIp = remoteSipIp;
  }

  public int getRemoteSipPort() {
    return remoteSipPort;
  }

  public void setRemoteSipPort(int remoteSipPort) {
    this.remoteSipPort = remoteSipPort;
  }

  public String getRemoteRtpIp() {
    return remoteRtpIp;
  }

  public void setRemoteRtpIp(String remoteRtpIp) {
    this.remoteRtpIp = remoteRtpIp;
  }

  public int getRemoteRtpPort() {
    return remoteRtpPort;
  }

  public void setRemoteRtpPort(int remoteRtpPort) {
    this.remoteRtpPort = remoteRtpPort;
  }

  public int getLocalRtpPort() {
    return localRtpPort;
  }

  public void setLocalRtpPort(int localRtpPort) {
    this.localRtpPort = localRtpPort;
  }

  public long getCreatedTime() {
    return createdTime;
  }

  public void setCreatedTime(long createdTime) {
    this.createdTime = createdTime;
  }

  public long getUpdatedTime() {
    return updatedTime;
  }

  public void setUpdatedTime(long updatedTime) {
    this.updatedTime = updatedTime;
  }

  public long getAckDeadline() {
    return ackDeadline;
  }

  public void setAckDeadline(long ackDeadline) {
    this.ackDeadline = ackDeadline;
  }

  public boolean isAckReceived() {
    return ackReceived;
  }

  public void setAckReceived(boolean ackReceived) {
    this.ackReceived = ackReceived;
  }

  public boolean isTerminated() {
    return terminated;
  }

  public void setTerminated(boolean terminated) {
    this.terminated = terminated;
  }

  public String getLast200Ok() {
    return last200Ok;
  }

  public void setLast200Ok(String last200Ok) {
    this.last200Ok = last200Ok;
  }

  public RtpUdpServer getRtpServer() {
    return rtpServer;
  }

  public void setRtpServer(RtpUdpServer rtpServer) {
    this.rtpServer = rtpServer;
  }

  public CodecSpec getSelectedCodec() {
    return selectedCodec;
  }

  public void setSelectedCodec(CodecSpec selectedCodec) {
    this.selectedCodec = selectedCodec;
  }

  public boolean isTelephoneEventSupported() {
    return telephoneEventSupported;
  }

  public void setTelephoneEventSupported(boolean telephoneEventSupported) {
    this.telephoneEventSupported = telephoneEventSupported;
  }

  public int getRemoteTelephoneEventPayloadType() {
    return remoteTelephoneEventPayloadType;
  }

  public void setRemoteTelephoneEventPayloadType(int remoteTelephoneEventPayloadType) {
    this.remoteTelephoneEventPayloadType = remoteTelephoneEventPayloadType;
  }

  public int getPtime() {
    return ptime;
  }

  public void setPtime(int ptime) {
    this.ptime = ptime;
  }
}

package nexus.io.sip.rtp.packet;

public class RtpPacket {

  private int version;
  private boolean padding;
  private boolean extension;
  private int csrcCount;
  private boolean marker;
  private int payloadType;
  private int sequenceNumber;
  private long timestamp;
  private long ssrc;
  private byte[] payload;

  public int getVersion() {
    return version;
  }

  public void setVersion(int version) {
    this.version = version;
  }

  public boolean isPadding() {
    return padding;
  }

  public void setPadding(boolean padding) {
    this.padding = padding;
  }

  public boolean isExtension() {
    return extension;
  }

  public void setExtension(boolean extension) {
    this.extension = extension;
  }

  public int getCsrcCount() {
    return csrcCount;
  }

  public void setCsrcCount(int csrcCount) {
    this.csrcCount = csrcCount;
  }

  public boolean isMarker() {
    return marker;
  }

  public void setMarker(boolean marker) {
    this.marker = marker;
  }

  public int getPayloadType() {
    return payloadType;
  }

  public void setPayloadType(int payloadType) {
    this.payloadType = payloadType;
  }

  public int getSequenceNumber() {
    return sequenceNumber;
  }

  public void setSequenceNumber(int sequenceNumber) {
    this.sequenceNumber = sequenceNumber;
  }

  public long getTimestamp() {
    return timestamp;
  }

  public void setTimestamp(long timestamp) {
    this.timestamp = timestamp;
  }

  public long getSsrc() {
    return ssrc;
  }

  public void setSsrc(long ssrc) {
    this.ssrc = ssrc;
  }

  public byte[] getPayload() {
    return payload;
  }

  public void setPayload(byte[] payload) {
    this.payload = payload;
  }
}

package nexus.io.sip.rtp.packet;

public class RtpPacketParser {

  public RtpPacket parse(byte[] data) {
    if (data == null || data.length < 12) {
      throw new IllegalArgumentException("rtp packet too short");
    }

    int b0 = data[0] & 0xFF;
    int b1 = data[1] & 0xFF;

    int version = (b0 >> 6) & 0x03;
    if (version != 2) {
      throw new IllegalArgumentException("unsupported rtp version: " + version);
    }

    boolean padding = ((b0 >> 5) & 0x01) == 1;
    boolean extension = ((b0 >> 4) & 0x01) == 1;
    int csrcCount = b0 & 0x0F;

    boolean marker = ((b1 >> 7) & 0x01) == 1;
    int payloadType = b1 & 0x7F;

    int sequenceNumber = ((data[2] & 0xFF) << 8) | (data[3] & 0xFF);

    long timestamp =
        ((long) (data[4] & 0xFF) << 24) |
        ((long) (data[5] & 0xFF) << 16) |
        ((long) (data[6] & 0xFF) << 8) |
        ((long) (data[7] & 0xFF));

    long ssrc =
        ((long) (data[8] & 0xFF) << 24) |
        ((long) (data[9] & 0xFF) << 16) |
        ((long) (data[10] & 0xFF) << 8) |
        ((long) (data[11] & 0xFF));

    int headerLen = 12 + csrcCount * 4;

    if (data.length < headerLen) {
      throw new IllegalArgumentException("invalid rtp header length");
    }

    if (extension) {
      if (data.length < headerLen + 4) {
        throw new IllegalArgumentException("invalid rtp extension header");
      }
      int extLenWords = ((data[headerLen + 2] & 0xFF) << 8) | (data[headerLen + 3] & 0xFF);
      headerLen += 4 + extLenWords * 4;
      if (data.length < headerLen) {
        throw new IllegalArgumentException("invalid rtp extension payload");
      }
    }

    int payloadLen = data.length - headerLen;
    if (padding) {
      int paddingCount = data[data.length - 1] & 0xFF;
      payloadLen -= paddingCount;
      if (payloadLen < 0) {
        throw new IllegalArgumentException("invalid rtp padding");
      }
    }

    byte[] payload = new byte[payloadLen];
    System.arraycopy(data, headerLen, payload, 0, payloadLen);

    RtpPacket packet = new RtpPacket();
    packet.setVersion(version);
    packet.setPadding(padding);
    packet.setExtension(extension);
    packet.setCsrcCount(csrcCount);
    packet.setMarker(marker);
    packet.setPayloadType(payloadType);
    packet.setSequenceNumber(sequenceNumber);
    packet.setTimestamp(timestamp);
    packet.setSsrc(ssrc);
    packet.setPayload(payload);
    return packet;
  }
}

package nexus.io.sip.rtp.packet;

public class RtpPacketWriter {

  public byte[] write(RtpPacket packet) {
    byte[] payload = packet.getPayload();
    if (payload == null) {
      payload = new byte[0];
    }

    byte[] out = new byte[12 + payload.length];

    int b0 = 0;
    b0 |= (2 & 0x03) << 6; // version=2
    if (packet.isPadding()) {
      b0 |= 1 << 5;
    }
    if (packet.isExtension()) {
      b0 |= 1 << 4;
    }
    b0 |= (packet.getCsrcCount() & 0x0F);

    int b1 = 0;
    if (packet.isMarker()) {
      b1 |= 1 << 7;
    }
    b1 |= (packet.getPayloadType() & 0x7F);

    out[0] = (byte) b0;
    out[1] = (byte) b1;

    int seq = packet.getSequenceNumber() & 0xFFFF;
    out[2] = (byte) ((seq >> 8) & 0xFF);
    out[3] = (byte) (seq & 0xFF);

    long ts = packet.getTimestamp() & 0xFFFFFFFFL;
    out[4] = (byte) ((ts >> 24) & 0xFF);
    out[5] = (byte) ((ts >> 16) & 0xFF);
    out[6] = (byte) ((ts >> 8) & 0xFF);
    out[7] = (byte) (ts & 0xFF);

    long ssrc = packet.getSsrc() & 0xFFFFFFFFL;
    out[8] = (byte) ((ssrc >> 24) & 0xFF);
    out[9] = (byte) ((ssrc >> 16) & 0xFF);
    out[10] = (byte) ((ssrc >> 8) & 0xFF);
    out[11] = (byte) (ssrc & 0xFF);

    System.arraycopy(payload, 0, out, 12, payload.length);
    return out;
  }
}

package nexus.io.sip.rtp.media;

public class AudioFrame {

  private short[] samples;
  private int sampleRate;
  private int channels;
  private long rtpTimestamp;

  public AudioFrame() {
  }

  public AudioFrame(short[] samples, int sampleRate, int channels, long rtpTimestamp) {
    this.samples = samples;
    this.sampleRate = sampleRate;
    this.channels = channels;
    this.rtpTimestamp = rtpTimestamp;
  }

  public short[] getSamples() {
    return samples;
  }

  public void setSamples(short[] samples) {
    this.samples = samples;
  }

  public int getSampleRate() {
    return sampleRate;
  }

  public void setSampleRate(int sampleRate) {
    this.sampleRate = sampleRate;
  }

  public int getChannels() {
    return channels;
  }

  public void setChannels(int channels) {
    this.channels = channels;
  }

  public long getRtpTimestamp() {
    return rtpTimestamp;
  }

  public void setRtpTimestamp(long rtpTimestamp) {
    this.rtpTimestamp = rtpTimestamp;
  }

  public int sampleCount() {
    return samples == null ? 0 : samples.length;
  }
}

package nexus.io.sip.rtp.media;

import nexus.io.sip.model.CallSession;

public interface MediaProcessor {
  AudioFrame process(AudioFrame input, CallSession session);
}

package nexus.io.sip.rtp.media;

import nexus.io.sip.model.CallSession;

public class EchoMediaProcessor implements MediaProcessor {

  @Override
  public AudioFrame process(AudioFrame input, CallSession session) {
    return input;
  }
}

package nexus.io.sip.rtp.codec;

public interface AudioCodec {

  String codecName();

  int payloadType();

  int sampleRate();

  short[] decode(byte[] payload);

  byte[] encode(short[] pcm16);
}

package nexus.io.sip.rtp.codec;

public class PcmuCodec implements AudioCodec {

  @Override
  public String codecName() {
    return "PCMU";
  }

  @Override
  public int payloadType() {
    return 0;
  }

  @Override
  public int sampleRate() {
    return 8000;
  }

  @Override
  public short[] decode(byte[] payload) {
    short[] out = new short[payload.length];
    for (int i = 0; i < payload.length; i++) {
      out[i] = ulawToLinear(payload[i]);
    }
    return out;
  }

  @Override
  public byte[] encode(short[] pcm16) {
    byte[] out = new byte[pcm16.length];
    for (int i = 0; i < pcm16.length; i++) {
      out[i] = linearToUlaw(pcm16[i]);
    }
    return out;
  }

  private short ulawToLinear(byte ulaw) {
    int u = ~ulaw & 0xFF;
    int sign = u & 0x80;
    int exponent = (u >> 4) & 0x07;
    int mantissa = u & 0x0F;
    int sample = ((mantissa << 3) + 0x84) << exponent;
    sample -= 0x84;
    return (short) (sign != 0 ? -sample : sample);
  }

  private byte linearToUlaw(short sample) {
    final int BIAS = 0x84;
    final int CLIP = 32635;

    int pcm = sample;
    int sign = (pcm >> 8) & 0x80;
    if (sign != 0) {
      pcm = -pcm;
    }
    if (pcm > CLIP) {
      pcm = CLIP;
    }

    pcm += BIAS;

    int exponent = 7;
    for (int expMask = 0x4000; (pcm & expMask) == 0 && exponent > 0; exponent--, expMask >>= 1) {
    }

    int mantissa = (pcm >> (exponent + 3)) & 0x0F;
    int ulaw = ~(sign | (exponent << 4) | mantissa) & 0xFF;
    return (byte) ulaw;
  }
}

package nexus.io.sip.rtp.codec;

public class PcmaCodec implements AudioCodec {

  @Override
  public String codecName() {
    return "PCMA";
  }

  @Override
  public int payloadType() {
    return 8;
  }

  @Override
  public int sampleRate() {
    return 8000;
  }

  @Override
  public short[] decode(byte[] payload) {
    short[] out = new short[payload.length];
    for (int i = 0; i < payload.length; i++) {
      out[i] = alawToLinear(payload[i]);
    }
    return out;
  }

  @Override
  public byte[] encode(short[] pcm16) {
    byte[] out = new byte[pcm16.length];
    for (int i = 0; i < pcm16.length; i++) {
      out[i] = linearToAlaw(pcm16[i]);
    }
    return out;
  }

  private short alawToLinear(byte alaw) {
    int a = alaw ^ 0x55;
    int sign = a & 0x80;
    int exponent = (a & 0x70) >> 4;
    int mantissa = a & 0x0F;

    int sample;
    if (exponent == 0) {
      sample = (mantissa << 4) + 8;
    } else {
      sample = ((mantissa << 4) + 0x108) << (exponent - 1);
    }

    return (short) (sign == 0 ? sample : -sample);
  }

  private byte linearToAlaw(short sample) {
    int pcm = sample;
    int sign;
    int exponent;
    int mantissa;
    int alaw;

    sign = (pcm & 0x8000) >> 8;
    if (sign != 0) {
      pcm = -pcm;
    }

    if (pcm > 32767) {
      pcm = 32767;
    }

    if (pcm >= 256) {
      exponent = 7;
      for (int expMask = 0x4000; (pcm & expMask) == 0 && exponent > 0; exponent--, expMask >>= 1) {
      }
      mantissa = (pcm >> (exponent + 3)) & 0x0F;
      alaw = (exponent << 4) | mantissa;
    } else {
      alaw = pcm >> 4;
    }

    alaw ^= (sign ^ 0x55);
    return (byte) alaw;
  }
}

package nexus.io.sip.rtp.server;

import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;

import nexus.io.sip.model.CallSession;
import nexus.io.sip.rtp.codec.AudioCodec;
import nexus.io.sip.rtp.codec.PcmaCodec;
import nexus.io.sip.rtp.codec.PcmuCodec;
import nexus.io.sip.rtp.media.AudioFrame;
import nexus.io.sip.rtp.media.EchoMediaProcessor;
import nexus.io.sip.rtp.media.MediaProcessor;
import nexus.io.sip.rtp.packet.RtpPacket;
import nexus.io.sip.rtp.packet.RtpPacketParser;
import nexus.io.sip.rtp.packet.RtpPacketWriter;
import nexus.io.sip.server.session.CallSessionManager;
import nexus.io.tio.core.Node;
import nexus.io.tio.core.udp.UdpPacket;
import nexus.io.tio.core.udp.intf.UdpHandler;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class RtpUdpHandler implements UdpHandler {

  private final int localPort;
  private final CallSessionManager sessionManager;

  private final RtpPacketParser rtpPacketParser = new RtpPacketParser();
  private final RtpPacketWriter rtpPacketWriter = new RtpPacketWriter();
  private final MediaProcessor mediaProcessor;

  private final AudioCodec pcmuCodec = new PcmuCodec();
  private final AudioCodec pcmaCodec = new PcmaCodec();

  public RtpUdpHandler(int localPort, CallSessionManager sessionManager) {
    this(localPort, sessionManager, new EchoMediaProcessor());
  }

  public RtpUdpHandler(int localPort, CallSessionManager sessionManager, MediaProcessor mediaProcessor) {
    this.localPort = localPort;
    this.sessionManager = sessionManager;
    this.mediaProcessor = mediaProcessor;
  }

  @Override
  public void handler(UdpPacket udpPacket, DatagramSocket socket) {
    try {
      CallSession session = sessionManager.getByLocalRtpPort(localPort);
      if (session == null || session.isTerminated()) {
        return;
      }

      Node remote = udpPacket.getRemote();
      byte[] data = udpPacket.getData();
      if (data == null || data.length < 12) {
        return;
      }

      RtpPacket in = rtpPacketParser.parse(data);

      // 更新远端 RTP 地址，适配首次学习或端口漂移
      if (session.getRemoteRtpIp() == null || session.getRemoteRtpIp().isEmpty()) {
        session.setRemoteRtpIp(remote.getIp());
      }
      if (session.getRemoteRtpPort() <= 0) {
        session.setRemoteRtpPort(remote.getPort());
      }

      // DTMF event 先忽略，不做 echo
      if (session.isTelephoneEventSupported() &&
          in.getPayloadType() == session.getRemoteTelephoneEventPayloadType()) {
        session.setUpdatedTime(System.currentTimeMillis());
        return;
      }

      AudioCodec codec = chooseCodec(session);
      if (codec == null) {
        return;
      }

      // 有些终端可能发来的 payload type 和协商结果不一致，先只按 session 选中 codec 解码
      short[] pcm = codec.decode(in.getPayload());
      AudioFrame inputFrame = new AudioFrame(pcm, codec.sampleRate(), 1, in.getTimestamp());

      AudioFrame outputFrame = mediaProcessor.process(inputFrame, session);
      if (outputFrame == null || outputFrame.getSamples() == null || outputFrame.getSamples().length == 0) {
        return;
      }

      byte[] outPayload = codec.encode(outputFrame.getSamples());

      RtpPacket out = new RtpPacket();
      out.setVersion(2);
      out.setPadding(false);
      out.setExtension(false);
      out.setCsrcCount(0);
      out.setMarker(false);
      out.setPayloadType(session.getSelectedCodec().getPayloadType());
      out.setSequenceNumber(session.nextSendSequence());
      out.setTimestamp(session.nextSendTimestamp(outputFrame.sampleCount()));
      out.setSsrc(session.getLocalSsrc());
      out.setPayload(outPayload);

      byte[] outBytes = rtpPacketWriter.write(out);

      DatagramPacket resp = new DatagramPacket(
          outBytes,
          outBytes.length,
          new InetSocketAddress(session.getRemoteRtpIp(), session.getRemoteRtpPort()));
      socket.send(resp);

      session.setUpdatedTime(System.currentTimeMillis());
    } catch (Exception e) {
      log.error("rtp handler error, localPort={}", localPort, e);
    }
  }

  private AudioCodec chooseCodec(CallSession session) {
    if (session.getSelectedCodec() == null) {
      return null;
    }

    String codecName = session.getSelectedCodec().getCodecName();
    if ("PCMU".equalsIgnoreCase(codecName)) {
      return pcmuCodec;
    }
    if ("PCMA".equalsIgnoreCase(codecName)) {
      return pcmaCodec;
    }
    return null;
  }
}

package nexus.io.sip.server.session;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import nexus.io.sip.model.CallSession;

public class CallSessionManager {

  private final Map<String, CallSession> sessions = new ConcurrentHashMap<>();

  public CallSession getByCallId(String callId) {
    if (callId == null) {
      return null;
    }
    return sessions.get(callId);
  }

  public CallSession getByLocalRtpPort(int localRtpPort) {
    for (CallSession session : sessions.values()) {
      if (session != null && session.getLocalRtpPort() == localRtpPort) {
        return session;
      }
    }
    return null;
  }

  public CallSession createOrUpdate(CallSession session) {
    if (session == null || session.getCallId() == null) {
      throw new IllegalArgumentException("call session or callId is null");
    }
    session.setUpdatedTime(System.currentTimeMillis());
    sessions.put(session.getCallId(), session);
    return session;
  }

  public void markAckReceived(String callId) {
    CallSession session = sessions.get(callId);
    if (session != null) {
      session.setAckReceived(true);
      session.setUpdatedTime(System.currentTimeMillis());
    }
  }

  public void markTerminated(String callId) {
    CallSession session = sessions.get(callId);
    if (session != null) {
      session.setTerminated(true);
      session.setUpdatedTime(System.currentTimeMillis());
    }
  }

  public void remove(String callId) {
    sessions.remove(callId);
  }

  public Map<String, CallSession> snapshot() {
    return Map.copyOf(sessions);
  }
}

package nexus.io.sip.rtp;

import java.net.SocketException;

import nexus.io.sip.rtp.media.MediaProcessor;
import nexus.io.sip.rtp.server.RtpUdpHandler;
import nexus.io.sip.server.session.CallSessionManager;
import nexus.io.tio.core.udp.UdpServer;
import nexus.io.tio.core.udp.UdpServerConf;

public class RtpUdpServer {

  private final int port;
  private final CallSessionManager sessionManager;
  private UdpServer udpServer;

  public RtpUdpServer(int port, CallSessionManager sessionManager) {
    this.port = port;
    this.sessionManager = sessionManager;
  }

  public void start() throws SocketException {
    RtpUdpHandler udpHandler = new RtpUdpHandler(port, sessionManager);
    start(udpHandler);
  }

  public void start(MediaProcessor mediaProcessor) throws SocketException {
    RtpUdpHandler udpHandler = new RtpUdpHandler(port, sessionManager, mediaProcessor);
    start(udpHandler);
  }

  public void start(RtpUdpHandler udpHandler) throws SocketException {
    UdpServerConf conf = new UdpServerConf(port, udpHandler, 5000);
    this.udpServer = new UdpServer(conf);
    this.udpServer.start();
  }

  public void stop() {
    if (udpServer != null) {
      udpServer.stop();
    }
  }

  public int port() {
    return port;
  }
}

package nexus.io.sip.rtp;

import nexus.io.sip.model.CallSession;
import nexus.io.sip.rtp.media.MediaProcessor;
import nexus.io.sip.server.session.CallSessionManager;

public class RtpServerManager {

  private final String localIp;
  private final RtpPortAllocator allocator;
  private final CallSessionManager sessionManager;

  public RtpServerManager(String localIp, CallSessionManager sessionManager) {
    this(localIp, new RtpPortAllocator(), sessionManager);
  }

  public RtpServerManager(String localIp, RtpPortAllocator allocator, CallSessionManager sessionManager) {
    this.localIp = localIp;
    this.allocator = allocator;
    this.sessionManager = sessionManager;
  }

  public CallSession allocateAndStart(CallSession session, MediaProcessor mediaProcessor) throws Exception {
    int rtpPort = allocator.allocate();
    RtpUdpServer rtpServer = new RtpUdpServer(rtpPort, sessionManager);
    rtpServer.start(mediaProcessor);

    session.setLocalRtpPort(rtpPort);
    session.setRtpServer(rtpServer);
    session.setUpdatedTime(System.currentTimeMillis());
    return session;
  }

  public void stopAndRelease(CallSession session) {
    if (session == null) {
      return;
    }

    try {
      if (session.getRtpServer() != null) {
        session.getRtpServer().stop();
      }
    } finally {
      if (session.getLocalRtpPort() > 0) {
        allocator.release(session.getLocalRtpPort());
      }
    }
  }

  public String getLocalIp() {
    return localIp;
  }
}

package nexus.io.sip.server.handler;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;

import nexus.io.aio.ByteBufferPacket;
import nexus.io.aio.Packet;
import nexus.io.sip.model.CallSession;
import nexus.io.sip.model.SipMessage;
import nexus.io.sip.model.SipRequest;
import nexus.io.sip.model.SipResponse;
import nexus.io.sip.parser.SipMessageEncoder;
import nexus.io.sip.parser.SipMessageParser;
import nexus.io.sip.parser.SipTcpFrameDecoder;
import nexus.io.sip.rtp.RtpServerManager;
import nexus.io.sip.rtp.media.MediaProcessor;
import nexus.io.sip.sdp.SdpAnswerBuilder;
import nexus.io.sip.sdp.SdpNegotiationResult;
import nexus.io.sip.sdp.SdpParser;
import nexus.io.sip.server.session.CallSessionManager;
import nexus.io.tio.core.ChannelContext;
import nexus.io.tio.core.Tio;
import nexus.io.tio.core.TioConfig;
import nexus.io.tio.server.intf.ServerAioHandler;

public class SipTcpServerHandler implements ServerAioHandler {

  private final String localIp;
  private final SipTcpFrameDecoder frameDecoder = new SipTcpFrameDecoder();
  private final SipMessageParser messageParser = new SipMessageParser();
  private final SipMessageEncoder messageEncoder = new SipMessageEncoder();
  private final CallSessionManager sessionManager;
  private final RtpServerManager rtpServerManager;
  private final MediaProcessor mediaProcessor;

  private final SdpParser sdpParser = new SdpParser();
  private final SdpAnswerBuilder sdpAnswerBuilder = new SdpAnswerBuilder();

  public SipTcpServerHandler(String localIp, CallSessionManager sessionManager, RtpServerManager rtpServerManager,
      MediaProcessor mediaProcessor) {
    this.localIp = localIp;
    this.sessionManager = sessionManager;
    this.rtpServerManager = rtpServerManager;
    this.mediaProcessor = mediaProcessor;
  }

  @Override
  public Packet decode(ByteBuffer buffer, int limit, int position, int readableLength, ChannelContext ctx)
      throws Exception {
    byte[] frame = frameDecoder.decode(buffer, readableLength, ctx);
    if (frame == null) {
      return null;
    }
    return new ByteBufferPacket(ByteBuffer.wrap(frame));
  }

  @Override
  public ByteBuffer encode(Packet packet, TioConfig tioConfig, ChannelContext ctx) {
    ByteBufferPacket p = (ByteBufferPacket) packet;
    ByteBuffer bb = p.getByteBuffer();
    if (bb.position() != 0) {
      bb.rewind();
    }
    return bb;
  }

  @Override
  public void handler(Packet packet, ChannelContext ctx) throws Exception {
    ByteBufferPacket p = (ByteBufferPacket) packet;
    ByteBuffer bb = p.getByteBuffer();

    byte[] bytes = new byte[bb.remaining()];
    bb.get(bytes);

    SipMessage msg = messageParser.parse(bytes);
    if (!(msg instanceof SipRequest)) {
      return;
    }

    SipRequest req = (SipRequest) msg;
    String method = req.getMethod();

    if ("INVITE".equalsIgnoreCase(method)) {
      handleInvite(req, ctx);
      return;
    }

    if ("ACK".equalsIgnoreCase(method)) {
      handleAck(req);
      return;
    }

    if ("BYE".equalsIgnoreCase(method)) {
      handleBye(req, ctx);
      return;
    }

    SipResponse resp = buildSimpleResponse(req, 200, "OK", null);
    send(ctx, resp);
  }

  private void handleInvite(SipRequest req, ChannelContext ctx) throws Exception {
    String callId = req.getHeader("Call-ID");
    CallSession exist = sessionManager.getByCallId(callId);

    if (exist != null && exist.getLast200Ok() != null) {
      sendRaw(ctx, exist.getLast200Ok());
      return;
    }

    SdpNegotiationResult negotiation = sdpParser.negotiate(req.getBody());
    if (!negotiation.isSuccess()) {
      SipResponse fail = buildSimpleResponse(req, 488, "Not Acceptable Here", null);
      send(ctx, fail);
      return;
    }

    String remoteIp = ctx.getClientNode() != null ? ctx.getClientNode().getIp() : null;
    int remotePort = ctx.getClientNode() != null ? ctx.getClientNode().getPort() : 0;

    String toTag = "java" + System.nanoTime();

    CallSession session = new CallSession();
    session.setCallId(callId);
    session.setFromTag(parseTag(req.getHeader("From")));
    session.setToTag(toTag);
    session.setTransport("TCP");
    session.setRemoteSipIp(remoteIp);
    session.setRemoteSipPort(remotePort);
    session.setCreatedTime(System.currentTimeMillis());
    session.setUpdatedTime(System.currentTimeMillis());
    session.setAckDeadline(System.currentTimeMillis() + 32000);

    session.setRemoteRtpIp(negotiation.getRemoteRtpIp());
    session.setRemoteRtpPort(negotiation.getRemoteRtpPort());
    session.setSelectedCodec(negotiation.getSelectedCodec());
    session.setTelephoneEventSupported(negotiation.isTelephoneEventSupported());
    session.setRemoteTelephoneEventPayloadType(negotiation.getRemoteTelephoneEventPayloadType());
    session.setPtime(negotiation.getPtime());

    rtpServerManager.allocateAndStart(session, mediaProcessor);

    SipResponse resp = buildInvite200Ok(req, session, negotiation);
    byte[] encoded = messageEncoder.encodeResponse(resp);
    String raw200 = new String(encoded, StandardCharsets.US_ASCII);

    session.setLast200Ok(raw200);
    sessionManager.createOrUpdate(session);

    Tio.send(ctx, new ByteBufferPacket(ByteBuffer.wrap(encoded)));
  }

  private void handleAck(SipRequest req) {
    String callId = req.getHeader("Call-ID");
    sessionManager.markAckReceived(callId);
  }

  private void handleBye(SipRequest req, ChannelContext ctx) throws Exception {
    String callId = req.getHeader("Call-ID");
    CallSession session = sessionManager.getByCallId(callId);

    SipResponse resp = buildSimpleResponse(req, 200, "OK", session != null ? session.getToTag() : null);
    send(ctx, resp);

    if (session != null) {
      rtpServerManager.stopAndRelease(session);
      sessionManager.markTerminated(callId);
    }
  }

  private void send(ChannelContext ctx, SipResponse response) {
    byte[] bytes = messageEncoder.encodeResponse(response);
    Tio.send(ctx, new ByteBufferPacket(ByteBuffer.wrap(bytes)));
  }

  private void sendRaw(ChannelContext ctx, String text) {
    byte[] bytes = text.getBytes(StandardCharsets.US_ASCII);
    Tio.send(ctx, new ByteBufferPacket(ByteBuffer.wrap(bytes)));
  }

  private SipResponse buildInvite200Ok(SipRequest req, CallSession session, SdpNegotiationResult negotiation) {
    SipResponse resp = new SipResponse();
    resp.setStatusCode(200);
    resp.setReasonPhrase("OK");

    copyIfPresent(req, resp, "Via");
    copyIfPresent(req, resp, "From");

    String to = req.getHeader("To");
    if (to != null && !to.toLowerCase().contains("tag=")) {
      to = to + ";tag=" + session.getToTag();
    }
    if (to != null) {
      resp.addHeader("To", to);
    }

    copyIfPresent(req, resp, "Call-ID");
    copyIfPresent(req, resp, "CSeq");
    resp.addHeader("Contact", "<sip:java@" + localIp + ":5060>");
    resp.addHeader("Content-Type", "application/sdp");

    String sdp = sdpAnswerBuilder.buildAnswer(localIp, session.getLocalRtpPort(), negotiation);
    resp.setBody(sdp.getBytes(StandardCharsets.US_ASCII));
    return resp;
  }

  private SipResponse buildSimpleResponse(SipRequest req, int code, String reason, String toTag) {
    SipResponse resp = new SipResponse();
    resp.setStatusCode(code);
    resp.setReasonPhrase(reason);

    copyIfPresent(req, resp, "Via");
    copyIfPresent(req, resp, "From");

    String to = req.getHeader("To");
    if (toTag != null && to != null && !to.toLowerCase().contains("tag=")) {
      to = to + ";tag=" + toTag;
    }
    if (to != null) {
      resp.addHeader("To", to);
    }

    copyIfPresent(req, resp, "Call-ID");
    copyIfPresent(req, resp, "CSeq");
    resp.setBody(new byte[0]);
    return resp;
  }

  private void copyIfPresent(SipRequest req, SipResponse resp, String headerName) {
    for (String v : req.getHeaders(headerName)) {
      resp.addHeader(headerName, v);
    }
  }

  private String parseTag(String headerValue) {
    if (headerValue == null) {
      return null;
    }

    String lower = headerValue.toLowerCase();
    int idx = lower.indexOf("tag=");
    if (idx < 0) {
      return null;
    }

    String sub = headerValue.substring(idx + 4);
    int semi = sub.indexOf(';');
    if (semi >= 0) {
      sub = sub.substring(0, semi);
    }
    return sub.trim();
  }
}