proxy.java

java ogg player library. for play back ogg audio

/* -*-mode:java; c-basic-offset:2; -*- */
/* JRoar -- pure Java streaming server for Ogg 
 *
 * Copyright (C) 2001,2002 ymnk, JCraft,Inc.
 *
 * Written by: 2001,2002 ymnk<[e-mail]>
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
 */

package com.jcraft.jroar;
import java.io.*;
import java.net.*;
import java.util.*;

import com.jcraft.jogg.*;

class Proxy extends Source implements Runnable{
  static final int BUFSIZE=4096*2;

//  private String source=null;
  private InputStream bitStream=null;

  private SyncState oy;
  private com.jcraft.jogg.Page og;
  
  private byte[] buffer=null;
  private int bytes=0;

  private Thread me=null;

  private int RETRY=3;
  int retry=RETRY;

  private long lasttime=0;

  Proxy(String mountpoint, String source){
    super(mountpoint);
    this.source=source;

HttpServer.source_connections++;

  }

  void init_ogg(){
    oy=new SyncState();
    og=new com.jcraft.jogg.Page();
    buffer=null;
    bytes=0;
    oy.init();
  }

  public void kick(){
    if(me!=null){
      if(System.currentTimeMillis()-lasttime>600000){
        //System.out.println("kick: stop!");
        stop();
        //System.out.println("kick: done!");
      }
      return;
    }
    if(mountpoint==null) return;
    me=new Thread(this);
    me.start();
  }

  public void run(){
    //if(me==null) return; 
    lasttime=System.currentTimeMillis();

    Vector http_header=new Vector();
    com.jcraft.jogg.Page[] pages=new com.jcraft.jogg.Page[10];
    int page_count=0;

    String _source=source;
    try{
      if(_source.startsWith("peercast://")){
        _source=PeerCast.getURL(_source);
      }
      else if(_source.endsWith(".pls")){
        _source=Pls.getURL(_source);
      }
      URL url=new URL(_source);
      URLConnection urlc=url.openConnection();

      setURLProperties(urlc);

      String foo;
      /*
      foo=urlc.getHeaderField(0); // HTTP/1.0 200 OK
      if(foo.indexOf(" 200 ")==-1){
        stop();
        return;
      }
      */

      int i=0;
      String s=null;
      String t=null;
      while(true){
        s=urlc.getHeaderField(i);
	t=urlc.getHeaderFieldKey(i);
	if(s==null)break;
	// System.out.println("header: "+t+": "+s);
	http_header.addElement((t==null?s:(t+": "+s)));
	i++;
      }

      int index=0;
      foo="jroar-source."+index+": ";
      i=0;
      for(;i<http_header.size();i++){
        s=(String)(http_header.elementAt(i));
        if(s.startsWith(foo)){
           index++;
           foo="jroar-source."+index+": ";
           i=0;
           continue;
	}
        break;
      }
      http_header.addElement(foo+source);

      bitStream=urlc.getInputStream();
    }
    catch(Exception ee){
      System.err.println(ee); 	    
      me=null;
//    drop();
      stop();
      return;
    }

    init_ogg();

    int serialno=-1;

    ByteArrayOutputStream _header=new ByteArrayOutputStream();
    byte[] header=null;

    retry=RETRY;

  loop:
    while(me!=null){
      boolean eos=false;
      header=null;
      while(!eos){
        int index=oy.buffer(BUFSIZE);
        buffer=oy.data;
        try{ bytes=bitStream.read(buffer, index, BUFSIZE); }
        catch(Exception e){
          System.err.println(e);
          bytes=-1;
          break;
        }
        if(bytes==-1)break;
        if(bytes==0)break;

        oy.wrote(bytes);

        lasttime=System.currentTimeMillis();

        try{Thread.sleep(1);}  // sleep for green thread.
        catch(Exception e){}

        while(!eos){
	  int result=oy.pageout(og);

	  if(result==0)break; // need more data
	  if(result==-1){ // missing or corrupt data at this page position
//	    System.err.println("Corrupt or missing data in bitstream; continuing...");
	  }
	  else{
            retry=RETRY;

//  	    if(serialno!=og.serialno()){
//              header=null;
//              serialno=og.serialno();
//	    }
          
            if((og.granulepos()==0)
               || (og.granulepos()==-1)          // hack for Speex
              ){
              if(header!=null){header=null;}
	      if(pages.length<=page_count){
		com.jcraft.jogg.Page[] foo=new com.jcraft.jogg.Page[pages.length*2];
		System.arraycopy(pages, 0, foo, 0, pages.length);
		pages=foo;
	      }
	      pages[page_count++]=og.copy();
	    }
            else{
              if(header==null){
		//parseHeader(pages, page_count);
		com.jcraft.jogg.Page foo;
		for(int i=0;i< page_count; i++){
		  foo=pages[i];
		  _header.write(foo.header_base, foo.header, foo.header_len);
		  _header.write(foo.body_base, foo.body, foo.body_len);
		}
                header=_header.toByteArray();
                _header.reset();
		page_count=0;
	      }
	    }

//          synchronized(listeners){  // In some case, c.write will block.
  	      int size=listeners.size();

              if(size==0){
  	        eos=true;

                stop();

                break;
	      }

              Client c=null;
              for(int i=0; i<size;){
	        try{
                  c=(Client)(listeners.elementAt(i));
                  c.write(http_header, header,
  			  og.header_base, og.header, og.header_len,
			  og.body_base, og.body, og.body_len);
		}
		catch(Exception e){
  	          c.close();
                  removeListener(c);
                  size--;
                  continue;
                }
                i++;
	      }
//  	    }
	    if(og.eos()!=0)eos=true;
	  }
        }
      }

      if(bytes==-1){
        retry--;
        if(retry>0){
          System.out.println("Connection to "+_source+" is dropped. Retry("+retry+")");
          header=null;
          serialno=-1;
          init_ogg();
          try { if(bitStream!=null)bitStream.close(); } 
          catch(Exception e) { System.out.println(e); }

          try{Thread.sleep(1000);}
          catch(Exception e) { }

          try{
            URL url=new URL(_source);
            URLConnection urlc=url.openConnection();

            setURLProperties(urlc);

            bitStream=urlc.getInputStream();
            continue;
	  }
	  catch(Exception e){
            retry=0;
	  }
	}
	else{
          stop();
	}
        break;
      }
    }
    stop();
  }

  private void setURLProperties(URLConnection urlc){
    if(HttpServer.myURL!=null){
      urlc.setRequestProperty("jroar-proxy", HttpServer.myURL+mountpoint);
      //System.out.println(HttpServer.myURL+mountpoint);
      if(JRoar.comment!=null)
        urlc.setRequestProperty("jroar-comment", JRoar.comment);
    }
  }

  void stop(){
    if(me!=null){
      if(oy!=null) oy.clear();
      try {
       if(bitStream!=null)bitStream.close();
      } 
      catch(Exception e) { }
      bitStream=null;
      me=null;
//    drop();
    }
    drop_clients();
  }

  void drop_clients(){
    Client c=null;
    synchronized(listeners){
      int size=listeners.size();
      for(int i=0; i<size;i++){
        c=(Client)(listeners.elementAt(i));
        try{ c.close();}
        catch(Exception e){}
      }
      listeners.removeAllElements();
    }
  }

  void drop(){
    drop_clients();
    super.drop();
  }

  public String toString(){
    return super.toString()+" me="+me+", bitStream="+bitStream;
  }
}