Changeset 260:4a9f47a6afbb

Timestamp:

04/30/2010 09:25:46 AM (4 months ago)

Author:

dcaoyuan

Branch:

default

Message:

Simplied the logic of SelectActor?

Location:

blogbird/src/main/scala/org/aiotrade/httpd

Files:

• ClientConnection.scala (modified) (2 diffs)
• HttpdActors.scala (modified) (2 diffs)
• SelectActor.scala (modified) (6 diffs)
• coroutine/HttpdRoutines.scala (modified) (3 diffs)

blogbird/src/main/scala/org/aiotrade/httpd/ClientConnection.scala

@@ -7 +7 @@
 import scala.actors.Actor
 import Debugger._
-
 
 /**
@@ -35 +34 @@
   def act = loop {
     react {
-      case Read(callback) =>
+      case Read(selector) =>
         try {
           val hasRequest = readMoreInput
-          callback(hasRequest)
+
+          // add back this connection to the read selector.
+          selector.addListener(this)
+
+          // Was enough input read to complete a request?
+          if (hasRequest) {
+            selector.asInstanceOf[{def produce(o: Request)}].produce(takeRequest)
+          }
         } catch {case ex: IOException => debugln(ex.getMessage)}
         
-      case Write(callback) =>
+      case Write(selector) =>
+        debugln(" ** Sending reply!")
         try {
           val finished = flushWriteBuffers
-          callback(finished)
+          if (finished) {
+            close
+          } else {
+          // add back this connection to the write selector.
+            selector.addListener(this)
+          }
         } catch {case ex: IOException => debugln(ex.getMessage)}
     }

blogbird/src/main/scala/org/aiotrade/httpd/HttpdActors.scala

@@ -66 +66 @@
 
   private object selectActor extends SelectActor[ClientConnection](SelectionKey.OP_READ) {
-
-    def fireEvent(key: SelectionKey, conn: ClientConnection) {
-      // Read whatever is available.
-      conn ! Read((hasRequest) => {
-          // Return this connection to the read selector.
-          addListener(conn)
-
-          // Was enough input read to complete a request?
-          if (hasRequest) {
-            produce(conn.takeRequest)
-          }
-        })
-    }
+    def produce(o: Request) = ConnectionManager.produce(o)
   }
 
@@ -127 +115 @@
   }
 
-  private object selectActor extends SelectActor[ClientConnection](SelectionKey.OP_WRITE) {
-
-    def fireEvent(key: SelectionKey, conn: ClientConnection) {
-      if (!conn.isOpen) {
-        conn.close
-      }
-
-      debugln(" ** Sending reply!")
-      conn ! Write(finished => {
-          if (!finished) {
-            addListener(conn)
-          } else {
-            conn.close
-          }
-        })
-    }
-  }
+  private object selectActor extends SelectActor[ClientConnection](SelectionKey.OP_WRITE)
 
   protected def consume(reply: Reply) {

blogbird/src/main/scala/org/aiotrade/httpd/SelectActor.scala

@@ -3 +3 @@
 import java.nio.channels.CancelledKeyException
 import java.nio.channels.SelectableChannel
-import java.nio.channels.SelectionKey
 import java.nio.channels.Selector
 import java.nio.channels.spi.SelectorProvider
@@ -10 +9 @@
 import scala.collection.JavaConversions._
 
-
-case class Read (callback: Boolean => Unit)
-case class Write(callback: Boolean => Unit)
+abstract class SelectionEvent[T <: WithSelectableChannel](sender: SelectActor[T])
+case class Read   [T <: WithSelectableChannel](sender: SelectActor[T]) extends SelectionEvent(sender)
+case class Write  [T <: WithSelectableChannel](sender: SelectActor[T]) extends SelectionEvent(sender)
+case class Unknown[T <: WithSelectableChannel](sender: SelectActor[T]) extends SelectionEvent(sender)
 
 abstract class SelectActor[T <: WithSelectableChannel](ops: Int) extends Actor {
-  def fireEvent(key: SelectionKey, listerner: T)
 
   private var done = false
+  // @Note tried using ConcurrentMap here, but no success
   private val newListeners = new LinkedBlockingQueue[T]
 
@@ -33 +33 @@
 
   def addListener(listener: T) {
-    newListeners.put(listener)
+    newListeners put listener
     selector.wakeup
   }
@@ -43 +43 @@
   def act {
     while (!done) {
-      selector.select // blocking call
+      selector.select // blocking call, you can think it's something like actor's receive
 
       val keys = selector.selectedKeys.iterator
@@ -52 +52 @@
         if (key.isValid) {
           val listener = key.attachment.asInstanceOf[T]
+          val event =
+            if (key.isReadable) {
+              Read(this)
+            } else if (key.isWritable) {
+              Write(this)
+            } else Unknown(this)
 
-          // Remove the key so we don't immediately loop around to
-          // race on the same connection.
+          // Remove the key so we don't immediately loop around to race on the same connection.
           key.cancel
 
-          fireEvent(key, listener)
+          // fire event
+          listener ! event
         }
       }
@@ -73 +79 @@
   }
 
+  /**
+   * override these methods to let me can do something according to
+   */
+  protected def postRead (condition: Boolean) {}
+  protected def postWrite(condition: Boolean) {}
+
 }
 

blogbird/src/main/scala/org/aiotrade/httpd/coroutine/HttpdRoutines.scala

@@ -63 +63 @@
 import java.nio.channels.SelectionKey
 import org.aiotrade.httpd.EmptyHostRouter
-import org.aiotrade.httpd.Write
 import org.aiotrade.httpd.HostRouter
 import org.aiotrade.httpd.Process
-import org.aiotrade.httpd.Read
 import org.aiotrade.httpd.Reply
 import org.aiotrade.httpd.Request
@@ -128 +126 @@
 
   private object selectActor extends SelectActor[ClientConnection](SelectionKey.OP_READ) {
-
-    def fireEvent(key: SelectionKey, conn: ClientConnection) {
-      // Read whatever is available.
-      conn ! Read((hasRequest) => {
-          // Return this connection to the read selector.
-          addListener(conn)
-
-          // Was enough input read to complete a request?
-          if (hasRequest) {
-            produce(conn.takeRequest)
-          }
-        })
-    }
+    def produce(o: Request) = ConnectionManager.produce(o)
   }
 
@@ -203 +189 @@
   }
 
-  private object selectActor extends SelectActor[ClientConnection](SelectionKey.OP_WRITE) {
-    val intersOpts = SelectionKey.OP_WRITE
-
-    def fireEvent(key: SelectionKey, conn: ClientConnection) {
-      if (!conn.isOpen) {
-        conn.close
-      }
-
-      debugln(" ** Sending reply!")
-      conn ! Write(finished => {
-          if (!finished) {
-            addListener(conn)
-          } else {
-            conn.close
-          }
-        })
-    }
-  }
+  private object selectActor extends SelectActor[ClientConnection](SelectionKey.OP_WRITE)
 
   def run {