java メッセージキューの処理 mainとは別のスレッドでA,Bを動かす

Question

前提・実現したいこと

クラスをAsanからBsanへキューを使ってメッセージを送る。
クラスをAsan,Bsanでくりかえしメッセージを送れるようにする。
Aさん,Bさんオブジェクトが,プロキシオブジェクトを介して通信するプログラムを作る。
実現するシーケンスは添付した画像の通りです。

これに
「proxyもAsan，Bsan同様にmainとは別のスレッドで動かす。」
という内容を追加したいのですがいまくいかないので質問させてもらいました。

■■な機能を実装中に以下のエラーメッセージが発生しました。

発生している問題・エラーメッセージ

エラーメッセージ

該当のソースコード

ソースコード

class Queue {
private String[] buf;
private int messagein = 0;
int start = 0;

public Queue(int size) {
    // 指定されたサイズでバッファを確保
    buf = new String[size];
}

public synchronized void writeMessage(String str) throws InterruptedException {
    // すでにメッセージがある場合
    while (messagein >= buf.length) {
        // notifyされるまで待機
        wait();
    }
    int end = (start + messagein) % buf.length;
    buf[end] = str;
    // メッセージが入った
    messagein++;
    notifyAll();
}

public synchronized String readMessage() throws InterruptedException {
    // メッセージが入るまで待つ
    while (messagein == 0) {
        // notifyされるまで待機
        wait();
    }
    String message = buf[start];
    start = (start + 1) % buf.length;
    // メッセージがなくなった
    messagein--;
    notifyAll();
    return message;
}

}

class Proxy {
Queue[] queue = new Queue[3];

public Proxy() {
    queue[0] = new Queue(10);
    queue[1] = new Queue(10);
    queue[2] = new Queue(10);
}

public void send(String source, String message) {
    try {
        if (source.equals("Aさん")) {
            queue[1].writeMessage(message);
            queue[0].writeMessage("100 Trying");
        } else {
            queue[0].writeMessage(message);
        }
    } catch (Exception e) {
        System.err.println(e.getMessage());
    }
}

public String recv(String source) {
    String ret = new String();

    try {
        if (source.equals("Aさん")) {
            ret = queue[0].readMessage();
        } else {
            ret = queue[1].readMessage();
        }
    } catch (Exception e) {
        System.err.println(e);
    }

    return ret;
}

}

class Asan implements Runnable {
Proxy proxy;
String name = "Aさん";

Asan(Proxy proxy) {
    this.proxy = proxy;
}

public void run() {
    try {
        String message = "INVITE";
        System.out.println(name + " send: " + message);
        proxy.send(name, message);
        Thread.sleep(1000);
        System.out.println(name + " recv: " + proxy.recv(name));
        Thread.sleep(1000);
        System.out.println(name + " recv: " + proxy.recv(name));
        Thread.sleep(5000);
        System.out.println(name + " recv: " + proxy.recv(name));
    } catch (Exception e) {
        System.err.println(e.getMessage());
    }
}

}

class Bsan implements Runnable {
Proxy proxy;
String name = "Bさん";

Bsan(Proxy proxy) {
    this.proxy = proxy;
}

public void run() {
    try {
        Thread.sleep(1000);
        System.out.println(name + " recv: " + proxy.recv(name));
        Thread.sleep(1000);
        String message = "180 Ringing";
        System.out.println(name + " send: " + message);
        proxy.send(name, message);
        Thread.sleep(3000);
        message = "200 OK";
        System.out.println(name + " send: " + message);
        proxy.send(name, message);
    } catch (Exception e) {
        System.err.println(e.getMessage());
    }
}

}

public class Main {
public static void main(String args[]) {
Proxy proxy = new Proxy();

    Asan sy1 = new Asan(proxy);
    Bsan sy2 = new Bsan(proxy);

    Thread th1 = new Thread(sy1);
    Thread th2 = new Thread(sy2);

    th1.start();
    th2.start();
}

}

試したこと

できた内容はソースコードのところまでです。

補足情報（FW/ツールのバージョンなど）

ここにより詳細な情報を記載してください。

Comments

[jimbe]

コードはファイル毎にマークダウン記法でご提示ください。
マークダウンにつきましてはヘルプをご参照ください。

[jimbe]

> ～ いまくいかないので ～
何が「うまく」いかないのでしょうか。ただ出来ませんでは丸投げになります。
どの部分でどの様な意図のコードがどうなってしまうのでしょうか。また、それに対してどの様な調査・対処を試みられて、どのような結果になったのでしょうか。
それらを「試したこと」に追記して頂きたく思います。

Answer 1

Proxy に受信用 Queue を作り、Proxy.send() はそのキューに追加、Proxyスレッドはそのキューで受信待ちをしていて、受信次第各送信先のキューに送信・・・でしょうか。

以下は、メッセージを"それっぽく"クラス化、Queueをジェネリクス化して送信先毎に Map 管理、Proxyスレッドの停止にもメッセージを使うようにする等しています。

package teratail_java.q370289;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;

class Queue<T> {
  private Object[] buf;
  private int messagein = 0;
  int start = 0;

  public Queue(int size) {
    // 指定されたサイズでバッファを確保
    buf = new Object[size];
  }
  public Queue() {
    this(10);
  }

  public synchronized void writeMessage(T message) throws InterruptedException {
    // すでにメッセージがある場合
    while (messagein >= buf.length) {
      // notifyされるまで待機
      wait();
    }
    int end = (start + messagein) % buf.length;
    buf[end] = message;
    // メッセージが入った
    messagein++;
    notifyAll();
  }

  public synchronized T readMessage() throws InterruptedException {
    // メッセージが入るまで待つ
    while (messagein == 0) {
      // notifyされるまで待機
      wait();
    }

    @SuppressWarnings("unchecked")
    T message = (T)buf[start];

    start = (start + 1) % buf.length;
    // メッセージがなくなった
    messagein--;
    notifyAll();
    return message;
  }
}

class Message {
  static final int CODE_INVITE = 0;
  static final int CODE_UNREGISTER = -999;

  final String src;
  final String dst;
  final int code;
  final String text;
  Message(String src, String dst, int code) {
    this(src, dst, code, "");
  }
  Message(String src, String dst, int code, String text) {
    this.src = src;
    this.dst = dst;
    this.code = code;
    this.text = text;
  }
  public String getTextWithCode() {
    return (code<=0?"":code+" ")+text;
  }
  @Override
  public String toString() {
    return "src='"+src+"', dst='"+dst+"', code="+code+", text='"+text+"'";
  }
}

class Proxy implements Runnable {
  static final String name = "Proxy";

  private Queue<Message> accept = new Queue<Message>();
  private Map<String,Queue<Message>> map = new HashMap<>(); //名前とキューの対応

  public void run() {
    try {
      while(true) {
        Message msg = accept.readMessage();
        if(msg.dst.equals(name)) {
          if(msg.code == Message.CODE_UNREGISTER) {
            map.remove(msg.src);
            if(map.isEmpty()) break; //利用者が居なくなったら終了
          }
        } else {
          getQueue(msg.dst).writeMessage(msg);
          if(msg.code == Message.CODE_INVITE) {
            getQueue(msg.src).writeMessage(new Message(name, msg.src, 100, "Trying"));
          }
        }
      }
    } catch (InterruptedException e) {
      e.printStackTrace(System.err);
    }
  }

  private Queue<Message> getQueue(String name) {
    synchronized(map) {
      Queue<Message> q = map.get(name);
      if(q == null) {
        q = new Queue<Message>();
        map.put(name, q);
      }
      return q;
    }
  }

  public void send(Message msg) throws InterruptedException {
    if(!msg.src.equals(name) && !msg.dst.equals(name)) { //Proxy[から|へ]は表示しない
      printLog(msg.src + " send: " + msg.getTextWithCode());
    }
    accept.writeMessage(msg);
  }

  public Message recv(String name) throws InterruptedException {
    Message msg = getQueue(name).readMessage();
    printLog(name + " recv: " + msg.getTextWithCode());
    return msg;
  }

  private void printLog(String log) {
    String time = DateTimeFormatter.ofPattern("HH:mm:ss ").format(LocalDateTime.now());
    System.out.println(time + log);
  }
}

class Asan implements Runnable {
  static final String name = "Aさん";

  private Proxy proxy;

  Asan(Proxy proxy) {
    this.proxy = proxy;
  }

  public void run() {
    try {
      proxy.send(new Message(name, Bsan.name, Message.CODE_INVITE, "INVITE"));
      Thread.sleep(1000);
      proxy.recv(name); //Trying?
      Thread.sleep(1000);
      proxy.recv(name); //Ringing?
      Thread.sleep(5000);
      proxy.recv(name); //OK?

      proxy.send(new Message(name, Proxy.name, Message.CODE_UNREGISTER));
    } catch (Exception e) {
      e.printStackTrace(System.err);
    }
  }
}

class Bsan implements Runnable {
  static final String name = "Bさん";

  private Proxy proxy;

  Bsan(Proxy proxy) {
    this.proxy = proxy;
  }

  public void run() {
    try {
      Thread.sleep(1000);
      Message recvMsg = proxy.recv(name); //INVITE?
      Thread.sleep(1000);
      proxy.send(new Message(name, recvMsg.src, 180, "Ringing"));
      Thread.sleep(3000);
      proxy.send(new Message(name, recvMsg.src, 200, "OK"));

      proxy.send(new Message(name, Proxy.name, Message.CODE_UNREGISTER));
    } catch (Exception e) {
      e.printStackTrace(System.err);
    }
  }
}

public class Main {
  public static void main(String args[]) throws InterruptedException {
    Proxy proxy = new Proxy();
    Asan sy1 = new Asan(proxy);
    Bsan sy2 = new Bsan(proxy);

    Thread threads[] = { new Thread(proxy), new Thread(sy1), new Thread(sy2) };

    for(Thread th : threads) th.start();
    for(Thread th : threads) th.join();
  }
}

実行結果

14:11:47 Aさん send: INVITE
14:11:48 Bさん recv: INVITE
14:11:48 Aさん recv: 100 Trying
14:11:49 Bさん send: 180 Ringing
14:11:49 Aさん recv: 180 Ringing
14:11:52 Bさん send: 200 OK
14:11:54 Aさん recv: 200 OK