Aufgabe11_GrafChat/src/ChatProgramm/model/Transmitter.java

/*
 * Click nbfs://nbhost/SystemFileSystem/Templates/Licenses/license-default.txt to change this license
 * Click nbfs://nbhost/SystemFileSystem/Templates/Classes/Class.java to edit this template
 */
package ChatProgramm.model;

import ChatProgramm.view.ChatView;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.io.UnsupportedEncodingException;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Flow;
import java.util.concurrent.Flow.Subscriber;
import java.util.concurrent.SubmissionPublisher;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 *
 * @author ahren
 */
public abstract class Transmitter implements Runnable, Subscriber<String>, TransmitterInterface {

    static final int timeout = 60000;
    private static final int PORT = 35000;

    private static Logger lg = Logger.getLogger("netz");

    protected Socket socket;
    protected BufferedReader reader;
    protected PrintWriter writer;
    
    private Nachricht nachricht;
    private String txtNachricht;
    private boolean laufend;
    private SubmissionPublisher<Nachricht> textPublisher;
    private ExecutorService eService;
    private String receivedString;
    private ChatView view;
    private ReceiveAdapter receiveAdapter;
    
    public Transmitter(ChatView view)
    {
      socket = new Socket();
      eService = null;
      receiveAdapter = new ReceiveAdapter(view);
      textPublisher = new SubmissionPublisher<>();
      this.view = view;
      addWertSubscription(receiveAdapter);
      nachricht = new Nachricht("");
    }
    
     public void addWertSubscription(Subscriber<Nachricht> subscriber)
    {
      textPublisher.subscribe(subscriber);
    }

    public abstract void connect() throws IOException;

    public void initIO() {
      try {
        lg.info("Initialisiere reader und writer");
        InputStream is = socket.getInputStream();
        OutputStream os = socket.getOutputStream();

        InputStreamReader isr = new InputStreamReader(is, "UTF-8");
        OutputStreamWriter osr = new OutputStreamWriter(os, "UTF-8");

        reader = new BufferedReader(isr);
        writer = new PrintWriter(osr);
        lg.info("Reader / Writer Initialisierung abgeschlossen");
        startempfangen();
        lg.info("Warte auf Nachricht");

      } catch (UnsupportedEncodingException ex) {
        Logger.getLogger(Transmitter.class.getName()).log(Level.SEVERE, null, ex);
      } catch (IOException ex) {
        Logger.getLogger(Transmitter.class.getName()).log(Level.SEVERE, null, ex);
      }
    }
    
    public void send(String txtNachricht){

        writer.println(txtNachricht);
        writer.flush();
        lg.info("Nachricht gesendet");
        nachricht.setNachricht("\nDu: " + txtNachricht);
        textPublisher.submit(nachricht);
    }
    public Nachricht receive(){
      try {
        txtNachricht = reader.readLine();
        if(!txtNachricht.isEmpty()){
          lg.info("Nachricht erhalten");
          nachricht.setNachricht("\nEr / Sie: " + txtNachricht);
          return nachricht;
        }

      } catch (IOException e) {
        throw new RuntimeException(e);
      }
      return nachricht;
    }
    
//    public void disconnect (){
//    in.close();
//    out.close();
//    s.close();
//    }
    
    @Override
    public void run() {
      while (true) {
        lg.info("Warte auf Nachricht");
        if(laufend) {
          nachricht = receive();
          if(!nachricht.getNachricht().isEmpty()){
              textPublisher.submit(nachricht);
          }
        }
        else{
          break;
        }
      }
    }


    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        throw new UnsupportedOperationException("Not supported yet."); // Generated from nbfs://nbhost/SystemFileSystem/Templates/Classes/Code/GeneratedMethodBody
    }

    @Override
    public void onNext(String item) {
        throw new UnsupportedOperationException("Not supported yet."); // Generated from nbfs://nbhost/SystemFileSystem/Templates/Classes/Code/GeneratedMethodBody
    }

    @Override
    public void onError(Throwable throwable) {
        throw new UnsupportedOperationException("Not supported yet."); // Generated from nbfs://nbhost/SystemFileSystem/Templates/Classes/Code/GeneratedMethodBody
    }

    @Override
    public void onComplete() {
        throw new UnsupportedOperationException("Not supported yet."); // Generated from nbfs://nbhost/SystemFileSystem/Templates/Classes/Code/GeneratedMethodBody
    }

  private void startempfangen()
  {
    synchronized (this){
    laufend = true;
    }
    if (eService == null){
        eService = Executors.newSingleThreadExecutor();
        eService.execute(this);
    }
    lg.info("Starte Chat");
  }
}