sandstormconfig.java

The Staged Event-Driven Architecture (SEDA) is a new design for building scalable Internet services.

    // Set command line values
    this.defaultInitArgs = new Hashtable();
    if (cmdLineArgs != null) {
      Enumeration e = cmdLineArgs.keys();
      while (e.hasMoreElements()) {
	String key = (String)e.nextElement();
	if (key.indexOf('.') != -1) {
	  putString(key, (String)cmdLineArgs.get(key));
	} else {
	  this.defaultInitArgs.put(key, (String)cmdLineArgs.get(key));
	}
      }
    }

    if (DEBUG) {
      System.err.println("DOING DUMP: -----------------------");
      root.dump();
      System.err.println("DONE WITH DUMP ---------------------");
    }

    // Get global init args
    configSection global = root.getSubsection("global");
    if (global != null) {
      global_initargs = global.getSubsection("initargs");
    }

    // Get stages
    if (DEBUG) System.err.println("Parsing stages");
    configSection stagesec = root.getSubsection("stages");
    if (stagesec != null) {
      for (int i = 0; i < stagesec.subsections.size(); i++) {
	configSection sec = (configSection)stagesec.subsections.elementAt(i);
	stageDescr descr = new stageDescr();
	descr.stageName = sec.getName();
	descr.className = sec.getVal("class");
	if (descr.className == null) throw new IOException("Missing class name in <stage> section of config file");
        if (DEBUG) System.err.println("Parsing stage "+descr.stageName);

      	descr.initargs = new Hashtable(1);

	// Add global args
	if (global_initargs != null) {
	  Enumeration e2 = global_initargs.getKeys();
	  while (e2.hasMoreElements()) {
	    String key = (String)e2.nextElement();
	    String val = global_initargs.getVal(key);
	    descr.initargs.put(key, val);
	  }
	}

	// Add stage-specific args
	configSection args = sec.getSubsection("initargs");
	if (args != null) {
	  Enumeration e2 = args.getKeys();
	  while (e2.hasMoreElements()) {
	    String key = (String)e2.nextElement();
	    String val = args.getVal(key);
	    descr.initargs.put(key,val);
	  }
	}

	// Add defaultInitArgs
	if (defaultInitArgs != null) {
	  Enumeration e2 = defaultInitArgs.keys();
	  while (e2.hasMoreElements()) {
	    String key = (String)e2.nextElement();
	    String val = (String)defaultInitArgs.get(key);
	    descr.initargs.put(key, val);
	  }
	}

	try {
	  String val = sec.getVal("queueThreshold");
	  if (val == null) descr.queueThreshold = -1;
	  else descr.queueThreshold = Integer.parseInt(val);
	} catch (NumberFormatException ne) {
	  descr.queueThreshold = -1;
	}
	if (DEBUG) System.err.println("Adding stage "+descr.stageName);
        stages.put(descr.stageName, descr);
      }
    }
  }


  // ----------------------------------------------------------------------

  // Convert an array of "key=value" strings to a Hashtable
  private Hashtable stringArrayToHT(String arr[]) throws IOException {
    if (arr == null) return null;
    Hashtable ht = new Hashtable(1);
    for (int i = 0; i < arr.length; i++) {
      StringTokenizer st = new StringTokenizer(arr[i], "=");
      String key;
      String val;
      try {
	key = st.nextToken();
	val = st.nextToken();
	while (st.hasMoreTokens()) val += "="+st.nextToken();
      } catch (NoSuchElementException e) {
	throw new IOException("Could not convert string '"+arr[i]+"' to key=value pair");
      }
      ht.put(key, val);
    }
    return ht;
  }

  // Internal class to represent configuration file format
  class configSection {
    private String secname;
    private StreamTokenizer tok;
    private Vector subsections;
    private Hashtable vals;

    private configSection() {
      subsections = new Vector(1);
      vals = new Hashtable(1);
    }

    configSection(Reader in) throws IOException {
      this();
      tok = new StreamTokenizer(in);
      tok.resetSyntax();
      tok.wordChars((char)0, (char)255);
      tok.whitespaceChars('\u0000', '\u0020');
      tok.commentChar('#');
      tok.eolIsSignificant(true);
      doRead();
    }

    private configSection(StreamTokenizer tok) throws IOException {
      this();
      this.tok = tok;
      tok.pushBack();
      tok.wordChars('0','9');
      doRead();
    }

    configSection(String name) {
      this();
      this.secname = name;
    }

    String getName() {
      return secname;
    }

    configSection getSubsection(String name) {
      for (int i = 0; i < subsections.size(); i++) {
	configSection sec = (configSection)subsections.elementAt(i);
	if (sec.getName().equals(name)) return sec;
      }
      return null;
    }

    void addSubsection(configSection subsec) {
      subsections.addElement(subsec);
    }

    // Return the string associated with key in this section
    // If not specified, null is returned
    String getVal(String key) {
      return (String)vals.get(key);
    }

    Enumeration getSubsections() {
      if (subsections == null) return null;
      return subsections.elements();
    }

    Enumeration getKeys() {
      return vals.keys();
    }

    int numKeys() {
      return vals.size();
    }

    Hashtable getVals() {
      return vals;
    }

    void putVal(String key, String val) {
      vals.put(key, val);
    }

    // Read next section name, parse recursively until we see the
    // end of that section
    private void doRead() throws IOException {
      String word, key, value;
      boolean read_secname = false;

      // Get initial section name
      word = nextWord();
      if (word.startsWith("<") && word.endsWith(">")) {
	secname = word.substring(1,word.length()-1);
      } else {
	throw new IOException("No section name found at line "+tok.lineno()+" of config file, read "+word);
      }

      boolean done = false;
      while (!done) {

	key = null;
	while (true) {
	  // Read key
	  word = nextWord();
	  if (word.startsWith("<") && word.endsWith(">")) {
	    String val = word.substring(1,word.length()-1);
	    if (val.equals("/"+secname)) {
	      // Done reading this section
	      done = true; break;
	    } else {
	      // Found a new section; recurse
	      configSection subsec = new configSection(tok);
	      if (getSubsection(subsec.getName()) != null) {
		throw new IOException("subsection "+subsec.getName()+" redefined at line "+tok.lineno()+" of config file");
	      }
	      if (vals.get(subsec.getName()) != null) {
		throw new IOException("subsection "+subsec.getName()+" conflicts with key "+subsec.getName()+" at line "+tok.lineno()+" of config file");
	      }

	      subsections.addElement(subsec);
	    }
	  } else {
	    key = word;
	    break;
	  }
	}

	if (done) break;

	// Read value
        word = nextLine();
        if (word.startsWith("<") && word.endsWith(">")) {
	  // Bad format: Should not have section tag here
	  throw new IOException("Unexpected section tag "+word+" on line "+tok.lineno()+" of config file");
	} else {
	  value = word;
	}

	if (key == null) throw new IOException("key is null at line "+tok.lineno()+" of config file");
	if (vals.get(key) != null) {
	  throw new IOException("key "+key+" redefined at line "+tok.lineno()+" of config file");
	}
	if (getSubsection(key) != null) {
	  throw new IOException("key "+key+" conflicts with subsection "+key+" at line "+tok.lineno()+" of config file");
	}
	if (key.indexOf(DELIM_CHAR) != -1) {
	  throw new IOException("key "+key+" may not contain character '"+DELIM_CHAR+"' at line "+tok.lineno()+" of config file");
	}
	vals.put(key,value);
      }

    }

    // Read next whitespace-delimited word from tok
    private String nextWord() throws IOException {
      while (true) {
	int type = tok.nextToken();
	switch (type) {

	  case StreamTokenizer.TT_EOL:
	    continue;

	  case StreamTokenizer.TT_EOF: 
	    throw new EOFException("EOF in config file");

	  case StreamTokenizer.TT_WORD:
	    if (DEBUG) System.err.println("nextWord returning "+tok.sval);
	    return tok.sval;

	  case StreamTokenizer.TT_NUMBER:
	    if (DEBUG) System.err.println("nextWord returning number");
	    return Double.toString(tok.nval);

	  default: 
	    continue;
	}
      }
    }

    // Read rest of line from tok
    private String nextLine() throws IOException {
      String line = new String("");
      boolean first = true;

      while (true) {
	switch (tok.nextToken()) {

	  case StreamTokenizer.TT_EOL:
	    if (DEBUG) System.err.println("nextLine returning "+line);
	    return line;

	  case StreamTokenizer.TT_EOF: 
	    throw new EOFException("EOF in config file");

	  case StreamTokenizer.TT_WORD:
	    if (first) {
	      line = tok.sval;
	      first = false;
            } else {
	      line += " "+tok.sval;
	    }
	    break;

	  case StreamTokenizer.TT_NUMBER:
	    if (first) {
	      line = Double.toString(tok.nval); 
	      first = false;
	    } else {
	      line += " "+Double.toString(tok.nval);
	    }
	    break;

	  default: 
	    continue;
	}
      }
    }

    // Debugging only
    void dump() {
      System.err.println("<"+secname+">");
      Enumeration e = vals.keys();
      while (e.hasMoreElements()) {
	String key = (String)e.nextElement();
	String val = (String)vals.get(key);
	System.err.println("   "+key+" "+val);
      }

      for (int i = 0; i < subsections.size(); i++) {
	configSection sec = (configSection)subsections.elementAt(i);
	sec.dump();
      }
      System.err.println("</"+secname+">");
    }

    public String toString() {
      return "configSection <"+secname+">";
    }
  }

  /** 
   * Internal class to preprocess special directives in the 
   * config file.
   */
  class directiveReader extends Reader {
    private Reader under, includedFile, markStream;
    private boolean markIsIncluded = false, closed = false;
    private boolean inComment = false;

    directiveReader(Reader under) throws IOException {
      this.under = under;
      if (!under.markSupported()) {
	throw new IOException("SandstormConfig: Internal error: directiveReader.under must support mark() -- contact [e-mail]");
      }
    }

    public int read() throws IOException {
      if (closed) throw new IOException("directiveReader is closed");
      if (includedFile != null) {
	int ret = includedFile.read();
	if (ret == -1) includedFile = null;
	else return ret;
      }

      boolean done = false;

      while (!done) {

	int c = under.read();

	// Ignore special directives inside of comments
	if (c == '#') {
	  inComment = true;
	}
	if (c == '\n') {
	  inComment = false;
	}

	if (!inComment && (c == '<')) {
	  under.mark(100);
	  if (under.read() == '!') {
	    // Process special directive; read until '>'
	    String directive = "<!";
	    char c1 = ' ';
	    while (c1 != '>') {
	      try {
		c1 = (char)under.read();
		if (c1 == -1) throw new IOException("End of file");
	      } catch (IOException ioe) {
		throw new IOException("SandstormConfig: Unterminated directive "+directive.substring(0,Math.min(directive.length(),10))+" in configuration file");
	      }
	      directive += c1;
	    }
	    if (DEBUG) System.err.println("Got special directive: "+directive);

	    if (directive.startsWith("<!include")) {
	      StringTokenizer st = new StringTokenizer(directive);
	      String dir = st.nextToken();
	      String fname = st.nextToken();
	      fname = fname.substring(0, fname.length()-1).trim();
	      if (DEBUG) System.err.println("Including file: "+fname);
	      includedFile = new directiveReader(new BufferedReader(new FileReader(fname)));
	      int ret = includedFile.read();
	      if (ret == -1) { 
		includedFile = null;
		continue;
	      } else {
		return ret;
	      }
	    } else {
	      throw new IOException("SandstormConfig: Unrecognized directive "+directive+" in config file");
	    }

	  } else {
	    // Got a '<' with no following '!'
	    under.reset();
	    return c;
	  }
	} else {
	  // Got something other than '<'
	  return c; 
	}
      }
      // Should never get here
      return -1;
    }

    public int read(char cbuf[]) throws IOException {
      return read(cbuf, 0, cbuf.length);
    }

    public int read(char cbuf[], int off, int len) throws IOException {
      int n = 0;
      for (int i = off; i < len; i++) {
	int c = read();
	if (cbuf[i] == -1) return n;
	cbuf[i] = (char)c;
	n++;
      }
      return n;
    }

    public long skip(long n) throws IOException {
      if (n < 0) throw new IllegalArgumentException("directiveReader.skip: n must be nonzero");
      long skipped = 0;
      for (long l = n; l >= 0; l--) {
	int c = read();
	if (c == -1) return skipped;
	skipped++;
      }
      return skipped;
    }

    public boolean ready() throws IOException {
      if (includedFile != null) return includedFile.ready();
      return under.ready();
    }

    public boolean markSupported() {
      return true;
    }

    public void mark(int readAheadLimit) throws IOException {
      if (includedFile != null) {
	markStream = includedFile;
	markIsIncluded = true;
      } else {
	markStream = under;
      }
      markStream.mark(readAheadLimit);
    }

    public void reset() throws IOException {
      markStream.reset();
      if (markIsIncluded) includedFile = markStream;
    }

    public void close() throws IOException {
      if (includedFile != null) includedFile.close();
      under.close();
      closed = true;
    }
  }

}