Aufgabe10_ChatProgramm/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.controller.Nachricht;
import ChatProgramm.controller.ReceiveAdapter;
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<Nachricht>, TransmitterInterface {

    static final int timeout = 30000;
    protected static final int PORT = 35000;
    protected static final String IP = "127.0.0.1";

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

    protected Socket socket;
    protected BufferedReader reader;
    protected PrintWriter writer;

    private SubmissionPublisher<Nachricht> textPublisher;
    private ExecutorService eService;
    private boolean laufend = false;

    private ChatView view;

    private ReceiveAdapter receiveAdapter;

    public Transmitter(ChatView view) {
        socket = new Socket();
        eService = null;
        this.view = view;
        textPublisher = new SubmissionPublisher<>();

        receiveAdapter = new ReceiveAdapter(view);
        addWertSubscription(receiveAdapter);
    }

    public abstract void connect() throws IOException;

    public void addWertSubscription(Subscriber<Nachricht> subscriber) {
        textPublisher.subscribe(subscriber);
    }

    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("Initialisierung abgeschlossen");

            start();

        } catch (IOException ex) {
            Logger.getLogger(Transmitter.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    public void start(){
        synchronized (this){
            laufend = true;
        }

        if (eService == null){
            eService = Executors.newSingleThreadExecutor();
            eService.execute(this);
        }

        lg.info("Starte Chat");
    }

    public void send(Nachricht nachricht) {
        lg.info("Nachricht wird gesendet");
        writer.println(nachricht.getNachricht());
        writer.flush();

        lg.info("Nachricht wird angezeigt");
        lg.info(nachricht.getNachricht());
        textPublisher.submit(nachricht);
    }

    public Nachricht recieve(){
        Nachricht nachricht = new Nachricht("");
        try {
            String txtNachricht = reader.readLine();
            if(!txtNachricht.isEmpty()){
                lg.info("Nachricht erhalten");
                nachricht.setNachricht(txtNachricht);
                return nachricht;
            }

        } catch (IOException e) {
            throw new RuntimeException(e);
        }


        return nachricht;
    }


    @Override
    public void run() {
        while (true) {
            lg.info("running");
            if(laufend) {
                Nachricht eingehendeNachricht = recieve();

                if(!eingehendeNachricht.getNachricht().isEmpty()){
                    textPublisher.submit(eingehendeNachricht);
                }

            }
            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 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
    }

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

}