async fetch

scalabl3 · Oct 25, 2011 · 7fe509e · 7fe509e
1 parent 19db247
commit 7fe509e
Show file tree

Hide file tree

Showing 2 changed files with 49 additions and 6 deletions.
diff --git a/imap/imap.go b/imap/imap.go
@@ -197,14 +197,18 @@ func (imap *IMAP) Examine(mailbox string) (*ResponseExamine, os.Error) {
 	return r, nil
 }
 
-func (imap *IMAP) Fetch(sequence string, fields []string) ([]*ResponseFetch, os.Error) {
+func formatFetch(sequence string, fields []string) string {
 	var fieldsStr string
 	if len(fields) == 1 {
 		fieldsStr = fields[0]
 	} else {
 		fieldsStr = "(" + strings.Join(fields, " ") + ")"
 	}
-	resp, err := imap.SendSync("FETCH %s %s", sequence, fieldsStr)
+	return fmt.Sprintf("FETCH %s %s", sequence, fieldsStr)
+}
+
+func (imap *IMAP) Fetch(sequence string, fields []string) ([]*ResponseFetch, os.Error) {
+	resp, err := imap.SendSync("%s", formatFetch(sequence, fields))
 	if err != nil {
 		return nil, err
 	}
@@ -220,6 +224,33 @@ func (imap *IMAP) Fetch(sequence string, fields []string) ([]*ResponseFetch, os.
 	return lists, nil
 }
 
+func (imap *IMAP) FetchAsync(sequence string, fields []string) (chan interface{}, os.Error) {
+	ch := make(chan interface{})
+	err := imap.Send(ch, formatFetch(sequence, fields))
+	if err != nil {
+		return nil, err
+	}
+
+	// Stream all responses to this message into outChan, and everything
+	// else into unsolicited.
+	outChan := make(chan interface{})
+	go func() {
+		for {
+			r := <-ch
+			switch r := r.(type) {
+			case *ResponseFetch:
+				outChan <- r
+			case *ResponseStatus:
+				outChan <- r
+				return
+			default:
+				imap.Unsolicited <- r
+			}
+		}
+	}()
+	return outChan, nil
+}
+
 // Repeatedly reads messages off the connection and dispatches them.
 func (imap *IMAP) readLoop() os.Error {
 	var msgChan chan interface{}

diff --git a/main.go b/main.go
@@ -132,11 +132,23 @@ func main() {
 
 		query := fmt.Sprintf("1:%d", examine.Exists)
 		vprintf("fetching %s", query)
-		fetches, err := im.Fetch(query, []string{"RFC822"})
+
+		ch, err := im.FetchAsync(query, []string{"RFC822"})
 		check(err)
-		for i, fetch := range fetches {
-			mbox.writeMessage(fetch.Rfc822)
-			fmt.Printf("%d\n", i)
+
+		i := 0
+	L:
+		for {
+			r := <-ch
+			switch r := r.(type) {
+			case *imap.ResponseFetch:
+				mbox.writeMessage(r.Rfc822)
+				fmt.Printf("%d\n", i)
+				i++
+			case *imap.ResponseStatus:
+				fmt.Printf("%v\n", r)
+				break L
+			}
 		}
 		readExtra(im)