/**
 *  Copyright © 2025, Luis Andrés Lange <https://javacomm.net>
 *
 *  This Source Code Form is subject to the terms of the Mozilla Public
 *  License, v. 2.0. If a copy of the MPL was not distributed with this
 *  file, You can obtain one at http://mozilla.org/MPL/2.0/.
 *
 *  ----------------------------------------------------------------------------
 *
 *  Exhibit B - "Incompatible With Secondary Licenses" Notice
 *
 *  This Source Code Form is "Incompatible With Secondary Licenses",
 *  as defined by the Mozilla Public License, v. 2.0.
 *
 *  In short:
 *  - This file may be used, modified, and distributed under MPL 2.0 only.
 *  - It may NOT be relicensed under GPL, LGPL, AGPL, or any other Secondary License.
 *
 *  Rationale:
 *  - Ensures that the code remains MPL-2.0.
 *  - Avoids legal conflicts with GPL-licensed libraries (e.g., VideoLAN).
 *  - Maximizes usability for commercial and security-critical applications.
 *
 */
package net.javacomm.client.telko;

import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import net.javacomm.client.websocket.WebsocketClient;
import net.javacomm.protocol.CONFERENCEAUDIO;
import net.javacomm.protocol.Command;
import net.javacomm.protocol.HEADER;
import net.javacomm.protocol.Protocol;



/**
 * Der Mikrofons Conference Consumer sendet Sprachpakete über einen Websocket.
 */
public class MicConferenceConsumer extends Thread {

  private static final Logger log = LogManager.getLogger(MicConferenceConsumer.class);

  private final ArrayBlockingQueue<byte[]> queue;
  private final WebsocketClient client;
  private final AtomicBoolean terminate = new AtomicBoolean(false);
  private final AtomicBoolean isMute = new AtomicBoolean(false);
  private final ReentrantLock lock = new ReentrantLock();
  private final Condition condition = lock.newCondition();

  private volatile String[] receiverSessions = new String[0];
  private String session;

  public MicConferenceConsumer(ArrayBlockingQueue<byte[]> queue, WebsocketClient client) {
    super("MicConferenceConsumer");
    this.queue = queue;
    this.client = client;
  }



  public void mute() {
    isMute.set(true);
  }



  @Override
  public void run() {
    String streamId = UUID.randomUUID().toString();

    while (!terminate.get()) {
      // mute-Logik mit Condition
      if (isMute.get()) {
        lock.lock();
        try {
          while (isMute.get() && !terminate.get()) {
            try {
              condition.await();
            }
            catch (InterruptedException e) {
              if (terminate.get()) {
                Thread.currentThread().interrupt();
                break;
              }
            }
          }
        }
        finally {
          lock.unlock();
        }
        if (terminate.get()) break;
      }

      try {
        byte[] audioBytes = queue.take(); // blockiert bis Daten vorhanden
        log.debug("MicConferenceConsumer: Paket mit {} Bytes aus Queue genommen.", audioBytes.length);

        if (terminate.get()) break;

        // Erstelle und sende CONFERENCEAUDIO
        CONFERENCEAUDIO conferenceAudio = new CONFERENCEAUDIO();
        conferenceAudio.setCommand(Command.CONFERENCEAUDIO);
        conferenceAudio.setHeader(HEADER.REQUEST);
        conferenceAudio.setDataset(Protocol.DATASET);
        conferenceAudio.setUserid(session);
        conferenceAudio.setStreamId(streamId);
        conferenceAudio.setReceiverSessions(receiverSessions);

        client.sendMessage(conferenceAudio);
        client.sendConferenceAudioBinary(streamId, audioBytes);
      }
      catch (InterruptedException e) {
        log.info("MicConferenceConsumer: Interrupted, Thread terminiert.");
        terminate.set(true);
        Thread.currentThread().interrupt();
        break;
      }
      catch (RuntimeException e) {
        log.error("MicConferenceConsumer: Unexpected error", e);
        terminate.set(true);
        break;
      }
    }

    log.info(MicConferenceConsumer.class.getSimpleName() + " hat terminiert.");
  }



  /**
   * Der Audiostream/Telefonat wird an diese Anwender gesendet. Jede Session wird
   * intern auf einen Anwender abgebildet.
   * 
   * @param sessions
   *                 alle Empfängersessions
   */
  public void setReceiverSessions(String[] sessions) {
    receiverSessions = sessions;
  }



  /**
   * Die Sprachnachricht stammt von diesem Absender.
   * 
   * @param session
   *                dieser Anwender
   */
  public void setSession(String session) {
    this.session = session;
  }



  public void terminate() {
    terminate.set(true);
    unmute(); // ggf. wartenden Thread wecken
    interrupt(); // falls queue.take() blockiert
  }



  /**
   * Das Mikrofon wurde auf Aufnahme gestellt.
   */
  public void unmute() {
    isMute.set(false);
    lock.lock();
    try {
      condition.signalAll(); // sicherer als signal()
    }
    finally {
      lock.unlock();
    }
  }
}
