go-xmpp2: comparison stream.go

equal deleted inserted replaced

-:6e2eea62ccca
+:c7f2edd25f4a
 // BUG(cjyar) Review all these *Client receiver methods. They should
 // probably either all be receivers, or none.
 func (cl *Client) readTransport(w io.Writer) {
+	defer fmt.Println("readTransport done")
 	defer tryClose(cl.socket, w)
 	cl.socket.SetReadTimeout(1e8)
 	p := make([]byte, 1024)
 	for {
 		if cl.socket == nil {
 		}
 	}
 }
 func (cl *Client) writeTransport(r io.Reader) {
+	defer fmt.Println("writeTransport done")
 	defer tryClose(r, cl.socket)
 	p := make([]byte, 1024)
 	for {
 		nr, err := r.Read(p)
 		if nr == 0 {
 	}
 }
 func readXml(r io.Reader, ch chan<- interface{},
 	extStanza map[string] func(*xml.Name) interface{}) {
+	defer fmt.Println("readXml done")
 	if Loglevel >= syslog.LOG_DEBUG {
 		pr, pw := io.Pipe()
 		go tee(r, pw, "S: ")
 		r = pr
 	}
 	defer tryClose(r, ch)
 	p := xml.NewParser(r)
+Loop:
 	for {
 		// Sniff the next token on the stream.
 		t, err := p.Token()
 		if t == nil {
 			if err != os.EOF {
 			if err != nil {
 				if Log != nil {
 					Log.Err("unmarshal stream: " +
 						err.String())
 				}
-				break
+				break Loop
 			}
 			ch <- st
 			continue
 		case "stream error", NsStream + " error":
 			obj = &streamError{}
 		err = p.Unmarshal(obj, &se)
 		if err != nil {
 			if Log != nil {
 				Log.Err("unmarshal: " + err.String())
 			}
-			break
+			break Loop
 		}
 		// If it's a Stanza, we try to unmarshal its innerxml
 		// into objects of the appropriate respective
 		// types. This is specified by our extensions.
 			if err != nil {
 				if Log != nil {
 					Log.Err("ext unmarshal: " +
 						err.String())
 				}
-				break
+				break Loop
 			}
 		}
 		// Put it on the channel.
 		ch <- obj
 		}
 	}
 }
 func (cl *Client) readStream(srvIn <-chan interface{}, cliOut chan<- Stanza) {
+	defer fmt.Println("readStream done")
+	defer close(cliOut)
 	defer tryClose(srvIn, cliOut)
 	handlers := make(map[string] func(Stanza) bool)
+Loop:
 	for {
 		select {
 		case h := <- cl.handlers:
 			handlers[h.id] = h.f
-		case x := <- srvIn:
+		case x, ok := <- srvIn:
-			if x == nil {
+			if !ok {
-				break
+				break Loop
 			}
 			send := false
 			switch obj := x.(type) {
 			case *stream:
 				handleStream(obj)
 // the app might inject something inappropriate into our negotiations
 // with the server. The control channel controls this loop's
 // activity.
 func writeStream(srvOut chan<- interface{}, cliIn <-chan Stanza,
 	control <-chan int) {
-	defer tryClose(srvOut, cliIn)
+	defer fmt.Println("writeStream done")
+	defer close(srvOut)
 	var input <-chan Stanza
+Loop:
 	for {
 		select {
 		case status := <- control:
 			switch status {
 			case 0:
 				input = nil
 			case 1:
 				input = cliIn
 			case -1:
-				break
+				break Loop
 			}
-		case x := <- input:
+		case x, ok := <- input:
+			if !ok {
+				fmt.Println("writeStream input closed")
+				break Loop
+			}
 			if x == nil {
 				if Log != nil {
 					Log.Notice("Refusing to send" +
 						" nil stanza")
 				}
 // Stanzas from the remote go up through a stack of filters to the
 // app. This function manages the filters.
 func filterTop(filterOut <-chan <-chan Stanza, filterIn chan<- <-chan Stanza,
 	topFilter <-chan Stanza, app chan<- Stanza) {
+	defer fmt.Println("filterTop done")
 	defer close(app)
+Loop:
 	for {
 		select {
 		case newFilterOut := <- filterOut:
 			if newFilterOut == nil {
 				if Log != nil {
 			filterIn <- topFilter
 			topFilter = newFilterOut
 		case data, ok := <-topFilter:
 			if !ok {
-				break
+				break Loop
 			}
 			app <- data
 		}
 	}
 }
 func filterBottom(from <-chan Stanza, to chan<- Stanza) {
+	defer fmt.Println("filterBottom done")
 	defer close(to)
 	for data := range(from) {
 		to <- data
 	}
 }
 func (cl *Client) handleStreamError(se *streamError) {
 	if Log != nil {
 		Log.Notice(fmt.Sprintf("Received stream error: %v", se))
 	}
-	cl.Close()
+	close(cl.Out)
 }
 func (cl *Client) handleFeatures(fe *Features) {
 	cl.Features = fe
 	if fe.Starttls != nil {
 	// Make the TLS connection available to the reader, and wait
 	// for it to signal that it's working again.
 	cl.socketSync.Add(1)
 	cl.socket = tls
 	cl.socketSync.Wait()
-	// Reset the read timeout on the (underlying) socket so the
-	// reader doesn't get woken up unnecessarily.
-	tcp.SetReadTimeout(0)
 	if Log != nil {
 		Log.Info("TLS negotiation succeeded.")
 	}
 	cl.Features = nil

branch	20120108-close
changeset 63	c7f2edd25f4a
parent 62	6e2eea62ccca
child 64	ac0639692317