例題04b ベースプロセス間のストリーム受け渡し

(1)概要

エージェントが持つベースプロセス間でストリームを受け渡す方法を示す。
(注意：ネームサーバNameServer.jarの起動が必要。)

(2)エージェント

エージェント名	ファイル名	説明
TimeSender	TimeSender.dash	ストリームを送信するエージェント
TimeSender	TimeSender.class	ベースプロセス
TimeReceiver	TimeReceiver.dash	ストリームを受信するエージェント
TimeReceiver	TimeReceiver.class	ベースプロセス

(2.1)TimeSender

●知識記述ファイル (TimeSender.dash)

(agent TimeSender
 
 (property
  (create :author "yu" :modifiedBy "taka")
 )
 
 (initial_facts
 )
 
 (knowledge
    (rule startup
      (Msg :performative __INIT_I)
      (Status :name ?name)
      -->
      (inspect nonstop)
      (loadBP baseProcess.dashSample.TimeSender)  // BP読み込み
      (startBP)                          // run()呼び出し
      (control startup ())               // startup()呼び出し
      (control setWindowTitle(?name))    // setWindowTitle()呼び出し
      (bind ?t (current-time))
      //キーとなるファクトを作成
      (make (key :name ?name :time ?t))
    )
    
    //パイプの作成とストリーム送信開始
    (rule sendStream
      (key :name ?name :time ?t) = ?arg
      -->
      (bind ?key (toString ?arg))
      //キーを用いてストリーム送受信用のパイプを作成し、
      //ストリーム送信開始
      (control startOutputStream(?key))
    )
    
    //キー要求を受け取り、キーを送る
    (rule receive-keyRequest
      (Msg :performative keyRequest :from ?sender ) = ?msg
      (key :name ?name :time ?t) = ?arg
      -->
      (bind ?key (toString ?arg))
      (send :performative sendKey :to ?sender :content (?key))
      (remove ?msg)
    )
  )
)

●ベースプロセス (TimeSender.java)
package baseProcess.dashSample; import java.util.*; import java.awt.*; import javax.swing.*; import javax.swing.border.*; import dash.*; import java.io.*; public class TimeSender extends JFrame implements DashBP { private DashAgent agent; JTextArea textarea; String key; BufferedWriter out; OutputStream os = null; //ループ用フラグ boolean flag = true; //同期用オブジェクト Object obj = new Object(); /** コンストラクタ。アクション(loadBP)実行時に呼ばれる。 */ public TimeSender() { super("TimeSender"); } /** アクション(loadBP)実行時に呼ばれる。*/ public void setAgent(DashAgent agent) { this.agent = agent; } /** アクション(startBP)実行時に呼ばれる */ public void run() { //出力ストリーム取得まで待つ synchronized (obj) { try { obj.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } //出力ストリーム取得後の処理 out = new BufferedWriter(new OutputStreamWriter(os)); while (flag) { Date now = new Date(); textarea.setText(now.toString()); try { out.write(now.toString() + "\n"); out.flush(); } catch (IOException e) { e.printStackTrace(); } try { Thread.sleep(1500); } catch (InterruptedException e) { e.printStackTrace(); } } } /** エージェントの終了時に呼ばれる */ public void finalizeBP() { flag = false; if (out != null) { //Pipeの切断処理を実行 agent.pm.close(key); } dispose(); } /** ルールstartup内で呼び出される */ public void startup() { setup(); // ときどきnullなことがある while (textarea == null) try { Thread.sleep(500); } catch (InterruptedException e) { } textarea.append("起動\n"); } /** ルールstartup内で呼び出される */ public void setWindowTitle(Object[] str1) { setTitle("sWin:" + str1[0]); } //Pipe作成、ストリーム送信開始 public void startOutputStream(Object[] str) { key = (String) str[0]; // キーを用いてPipeの作成処理を実行 agent.pm.create(key); synchronized (obj) { // キーを用いて出力ストリームを取得 os = agent.pm.getOutputStream(key); obj.notify(); } } /** ウィンドウを作る */ private void setup() { textarea = new JTextArea(6, 20); JScrollPane sc = new JScrollPane(textarea); sc.setBorder(new TitledBorder("Output")); getContentPane().add(sc, BorderLayout.CENTER); pack(); setVisible(true); } }

(2.2)TimeReceiver

●知識記述ファイル (TimeReceiver.dash)

(agent TimeReceiver
 
 (property
  (create :author "yu" :modifiedBy "taka")
 )
 
 (initial_facts
 )
 
  (knowledge
    (rule startup
      (Msg :performative __INIT_I)
      (Status :name ?name)
      -->
      (inspect nonstop)
      (loadBP baseProcess.dashSample.TimeReceiver)  // BP読み込み
      (startBP)                          // run()呼び出し
      (control startup ())
    )
 
    // ベースプロセスからのイベントを受け取り、
    // TimeSenderにキーを要求する
    (rule event-handling
      (Msg :performative __Event :content (?string)) = ?msg
      -->
      (bind ?agents (lookup :rname TimeSender))
      (bind ?agent (shift ?agents))
      (send :performative keyRequest :to ?agent:name :content ())
      (remove ?msg)
     )
    
    // TimeSenderから、キーを受け取る
    (rule receive-sendKey
     (Msg :performative sendKey :from ?sender :content (?key)) = ?msg
     -->
     //キーを用いてストリーム受信を開始
     (control startInputStream(?key))
     (remove ?msg)
    )
  )
)

●ベースプロセス (TimeReceiver.java)
package baseProcess.dashSample; import java.awt.*; import java.awt.event.*; import javax.swing.*; import javax.swing.border.*; import dash.*; import java.io.*; public class TimeReceiver extends JFrame implements DashBP, ActionListener { private DashAgent agent; JTextArea textarea; JButton startbutton; JButton stopbutton; BufferedReader in; InputStream is = null; //ループ用フラグ boolean flag = true; //同期用オブジェクト Object obj = new Object(); /** コンストラクタ。アクション(loadBP)実行時に呼ばれる。 */ public TimeReceiver() { super("TimeReceiver"); } /** アクション(loadBP)実行時に呼ばれる。*/ public void setAgent(DashAgent agent) { this.agent = agent; } /** アクション(startBP)実行時に呼ばれる * * */ public void run() { //入力ストリーム取得まで待つ synchronized (obj) { try { obj.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } //入力ストリーム取得後の処理 in = new BufferedReader(new InputStreamReader(is)); try { while (flag) { String ss = in.readLine(); if (ss != null) textarea.append(ss + "\n"); } } catch (IOException e) { finalizeBP(); //System.out.println("Pipeの接続が切れました"); } } //ストリーム受信開始 public void startInputStream(Object[] str) { synchronized (obj) { //入力ストリームを取得 is = agent.pm.getInputStream((String) str[0]); obj.notify(); } } /** エージェントの終了時に呼ばれる */ public void finalizeBP() { flag = false; if (in != null) { try { in.close(); } catch (IOException e) { e.printStackTrace(); } } dispose(); } /** startボタンを押し、エージェントにイベントを上げる */ public void actionPerformed(ActionEvent ae) { //System.out.println(((JButton)ae.getSource()).getText()); if (agent != null) { if (((JButton) ae.getSource()).getText().equals("Start")) { agent.raiseEvent("start"); } else if (((JButton) ae.getSource()).getText().equals("Finish")) { dispose(); } } } /** ルールstartup内で呼び出される */ public void startup() { setup(); // ときどきnullなことがある while (textarea == null) try { Thread.sleep(500); } catch (InterruptedException e) { } textarea.append("起動\n"); } /** ウィンドウを作る */ private void setup() { textarea = new JTextArea(6, 20); JScrollPane sc = new JScrollPane(textarea); sc.setBorder(new TitledBorder("Input")); startbutton = new JButton("Start"); startbutton.addActionListener(this); stopbutton = new JButton("Finish"); stopbutton.addActionListener(this); JPanel panel = new JPanel(); panel.add(startbutton); panel.add(stopbutton); GridLayout layout = new GridLayout(1, 2); panel.setLayout(layout); getContentPane().add(sc, BorderLayout.CENTER); getContentPane().add(panel, BorderLayout.SOUTH); pack(); setVisible(true); } }

(3)処理の流れ

①TimeSenderエージェント、TimeReceiverエージェントを起動する

②TimeSenderエージェントは、キーとなるファクトを生成し、そのキーを用いてストリーム送受信用のパイプを作成する（ agent.pm.create(key) ）

③TimeSenderエージェントは、キーを用いて出力ストリームを取得（ os = agent.pm.getOutputStream(key) ）し、ストリームを流す

④TimeReceiverエージェントは、ベースプロセスから受信開始のイベントを受け取ると、TimeSenderエージェントに対しキー要求メッセージを投げ、キーをもらう

⑤TimeReceiverエージェントは、キーをもとに入力ストリームを取得（ is = agent.pm.getInputStream(key) ）し、ストリームを受信する