jxtacast.java

基于jxta的局域网P2P文件共享,可以实现局域网中的文件p2p共享,实现文件快速传输及交流

      pipeAdvt = (PipeAdvertisement)AdvertisementFactory.newAdvertisement(
          PipeAdvertisement.getAdvertisementType());

      id = (PipeID)IDFactory.newPipeID(group.getPeerGroupID());
      pipeAdvt.setPipeID(id);
      pipeAdvt.setName(getBackChannelPipeName());
      pipeAdvt.setType(PipeService.UnicastType);

      try {
          disco.publish(pipeAdvt);
          disco.remotePublish(pipeAdvt);
          privInputPipe = pipeServ.createInputPipe(pipeAdvt, this);
      } catch (Exception e) {
          e.printStackTrace();
          return false;
      }

      return true;
  }


  /** Return the name used in advertisement for our "back channel" input pipe.
   *  The string contains a known prefix that can be used for discovery,
   *  plus our peer name and ID.
   */
  public String getBackChannelPipeName() {

      // Use a complex delimiter to mark off the peer name and ID.
      // We need to parse this string later, so we need something that's
      // unlikely to appear in a peer name.  (A simple period is too risky.)
      //
      String name = getBackChannelPipePrefix() + DELIM +
                    myPeer.getName()           + DELIM +
                    myPeer.getPeerID().toString();

      return name;
  }


  /** Return the prefix used in the name of our "back channel" input pipe.
   *  This prefix can be used with advertisement discovery to narrow the
   *  discovery results to peers using JxtaCast with your application.
   */
  public String getBackChannelPipePrefix() {

      return "FileCastBackChannel." + castName;
  }


  /** Extract the peer name from the given pipe advertisement name.
   */
  public static String getPeerNameFromBackChannelPipeName(String pipeName) {

      // The peer name is located between the first and second delimiters.
      int start = pipeName.indexOf(DELIM);
      if (start < 0)
          return null;

      int end = pipeName.indexOf(DELIM, start + 1);
      if (end < 0)
          return null;

      // Extract the peer name.
      start += DELIM.length();
      if (start > end)
          return null;
      return pipeName.substring(start, end);
  }


  /** Extract the peer ID from the given pipe advertisement name.
   */
  public static String getPeerIdFromBackChannelPipeName(String pipeName) {

      // The peer ID is located after the second delimiter.
      int pos = pipeName.indexOf(DELIM);
      if (pos < 0)
          return null;
      pos = pipeName.indexOf(DELIM, ++pos);
      if (pos < 0)
          return null;

      return pipeName.substring(pos + DELIM.length());
  }


  /**
   *  Send a Message down the output pipe.
   */
  public synchronized void sendMessage(Message msg) {

      try {
          outputPipe.send(msg);
      }
      catch (IOException e) {
          e.printStackTrace();
      }
  }


  /**
   * Receive messages from the input pipe.
   *
   * @param event PipeMsgEvent the event that contains our message.
   */
  public synchronized void pipeMsgEvent(PipeMsgEvent event) {

      Message msg = event.getMessage();
      try {
          // logMsg("Message received: " + getMsgString(msg, MESSAGETYPE));

          // Determine the message type, and dispatch it.
          String msgType = getMsgString(msg, MESSAGETYPE);
          if (msgType == null) {
              logMsg("Error: message received with no MESSAGETYPE.");
              return;
          }
          if (msgType.equals(MSG_CHAT))
              receiveChatMsg(msg);
          else
              receiveFileMsg(msg);

      } catch (Exception e) {
          e.printStackTrace();
      }
  }


  /**
   * Receive a file transfer message.
   *
   * @param msg a file transfer message.
   */
  public synchronized void receiveFileMsg(Message msg) {

      try {
          String msgType = getMsgString(msg, MESSAGETYPE);
          if (msgType == null)
              return;

          // Check for a wrangler in the hash table, to see if we've
          // already started processing this file.  If not, create a
          // new FileWrangler to handle it.  (But only if it's an a
          // MSG_FILE message from the original sender.)
          //
          FileWrangler wrangler = (FileWrangler)wranglers.get(getMsgString(msg, FILEKEY));
          if (wrangler == null  &&  msgType.equals(MSG_FILE)) {
              wrangler = new InputFileWrangler(this, msg);
              wranglers.put(wrangler.key, wrangler);
          }
          if (wrangler == null)
              logMsg("Unable to obtain wrangler for message.  Msg type: " +
                  msgType + "  key: " + getMsgString(msg, FILEKEY));
          else
              wrangler.processMsg(msg);


      } catch (Exception e) {
          e.printStackTrace();
      }
  }


  /**
   * Receive a chat message.
   *
   * @param msg a chat message.
   */
  public synchronized void receiveChatMsg(Message msg) {

      try {
          String sender   = getMsgString(msg, SENDERNAME);
          String caption  = getMsgString(msg, CAPTION);

          logMsg(sender + " : " + caption);

          // FIXME - Send the chat message to any registered listeners.
      } catch (Exception e) {
          e.printStackTrace();
      }
  }


  /**
   *  Send a chat message out to the group members.
   *
   *  @param  text   the message text.
   */
  public synchronized void sendChatMsg(String text) {
      try {
          // Create a message, fill it with our standard headers.
          Message msg = new Message();
          setMsgString(msg, MESSAGETYPE, MSG_CHAT);
          setMsgString(msg, SENDERNAME,  myPeer.getName());
          setMsgString(msg, SENDERID,    myPeer.getPeerID().toString());
          setMsgString(msg, VERSION,     version);
          setMsgString(msg, CAPTION,     text);

          sendMessage(msg);

      } catch (Exception ex) {
          ex.printStackTrace();
      }
  }


  /**
   *  Send a file out to the group members.
   *
   *  @param  file       the file to send.
   *  @param  caption    description of the file (optional)
   */
  public synchronized void sendFile(File file, String caption) {

      // Create a wrangler to handle the file transfer, and then queue it
      // in our sendFileQueue Vector.  Another thread will retrieve it
      // from the queue and start the send process.  (We don't want to
      // hang up the GUI while the file is loading.)
      //
      OutputFileWrangler wrangler = new OutputFileWrangler(this, file, caption);
      sendFileQueue.add(wrangler);
  }


  /** Worker thread.  Call functions to perform time-intensive operations that
   *  would bog down the main thread.
   */
  public void run() {

      while (true) {
          try {Thread.sleep(trailBossPeriod);} catch (InterruptedException e) {}

          synchronized (wranglers) {

              checkFileWranglers();
              checkSendFileQueue();
          }
      }
  }


  /** Loop thru our current collection of FileWrangler objects, and call
   *  bossCheck() for each one.  This gives them a chance to perform any
   *  needed tasks.
   *
   *  We keep wranglers stored in our Hashtable for awhile after we've finished
   *  sending or receiving the file. They're available to respond to requests
   *  from other peers for file blocks that they are missing.
   *
   *  In response to the bossCheck() call, wranglers that have been inactive
   *  for a long time will remove themselves from the collection.  Input
   *  wranglers that are missing file blocks will request them.
   */
  protected void checkFileWranglers() {

      Enumeration elements = wranglers.elements();;
      FileWrangler wrangler;

      while (elements.hasMoreElements()) {
          wrangler = (FileWrangler)elements.nextElement();
          wrangler.bossCheck();
      }
  }


  /**
   * Outgoing file send operations are queued by the main thread.  This function
   * is called by the worker thread.  It reads them from the queue, and triggers
   * the file load and send process.
   */
  protected void checkSendFileQueue() {

      OutputFileWrangler wrangler = null;

      // Yank the first wrangler from the queue, if there is one.
      // Put it in our collection of active wranglers, and start the
      // file load and send process.
      //
      if (sendFileQueue.isEmpty())
          return;
      wrangler = (OutputFileWrangler)sendFileQueue.remove(0);
      if (wrangler != null) {
          wranglers.put(wrangler.key, wrangler);
          wrangler.sendFile();
          wrangler = null;
      }
  }


  /**
   * Register a JxtaCastEventListener.  Listeners are sent progress events while
   * sending and receiving files.
   */
  public synchronized void addJxtaCastEventListener(JxtaCastEventListener listener) {

      // Add the listener to our collection, unless we already have it.
      if (!jcListeners.contains(listener))
          jcListeners.addElement(listener);
  }


  /**
   * Un-register a JxtaCastEventListener.
   */
  public synchronized void removeJxtaCastEventListener(JxtaCastEventListener listener) {

      jcListeners.removeElement(listener);
  }


  /**
   * Send a JxtaCastEvent to all registered listeners.
   */
  protected void sendJxtaCastEvent(JxtaCastEvent e) {

      JxtaCastEventListener listener = null;

      Enumeration elements = jcListeners.elements();
      while (elements.hasMoreElements()) {
          listener = (JxtaCastEventListener)elements.nextElement();
          listener.jxtaCastProgress(e);
      }
  }


  public static void setMsgString(Message msg, String name, String str) {

      msg.replaceMessageElement(new StringMessageElement(name, str, null));
  }


  public static String getMsgString(Message msg, String name) {

      MessageElement elem = msg.getMessageElement(name);
      if (elem == null)
          return null;

      return elem.toString();
  }
}



/**
* Base class for processing a file.
*
* Files are split up and sent in blocks of data.  This class gathers
* the data blocks for a file as they come in, and writes the file
* to disk when it is complete.
*
* It's ok for blocks to arrive out of order, and ok for duplicate blocks
* to arrive.
*/
abstract class FileWrangler {

  public String key;    // Unique identifier for this file transfer.

  JxtaCast jc;          // The parent JxtaCast obj.

  String sender;        // Sender's peer name.