yuxh/gearman-go
1
0
Fork 1
Code Issues Pull Requests Releases Wiki Activity

type assertion with *net.OpError

Browse Source
This commit is contained in:
Xing Xing 2014-03-03 14:45:35 +08:00
parent 45a9d7c3e5
commit f880354a61
2 changed files with 15 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
client/client.go
View File

@ -4,7 +4,6 @@ package client
import ( import (
"bufio" "bufio"
"io"
"net" "net"
"sync" "sync"
) )
@ -62,9 +61,6 @@ func (client *Client) read(length int) (data []byte, err error) {
// read until data can be unpacked // read until data can be unpacked
for i := length; i > 0 || len(data) < minPacketLength; i -= n { for i := length; i > 0 || len(data) < minPacketLength; i -= n {
if n, err = client.rw.Read(buf); err != nil { if n, err = client.rw.Read(buf); err != nil {
if err == io.EOF {
err = ErrLostConn
}
return return
} }
data = append(data, buf[0:n]...) data = append(data, buf[0:n]...)
@ -83,10 +79,16 @@ func (client *Client) readLoop() {
ReadLoop: ReadLoop:
for client.conn != nil { for client.conn != nil {
if data, err = client.read(bufferSize); err != nil { if data, err = client.read(bufferSize); err != nil {
client.err(err) if opErr, ok := err.(*net.OpError); ok {
if err == ErrLostConn { if opErr.Timeout() {
client.err(err)
}
if opErr.Temporary() {
continue
}
break break
} }
client.err(err)
// If it is unexpected error and the connection wasn't // If it is unexpected error and the connection wasn't
// closed by Gearmand, the client should close the conection // closed by Gearmand, the client should close the conection
// and reconnect to job server. // and reconnect to job server.

23
worker/agent.go
View File

@ -2,9 +2,7 @@ package worker
import ( import (
"bufio" "bufio"
"io"
"net" "net"
"strings"
"sync" "sync"
) )
@ -54,7 +52,13 @@ func (a *agent) work() {
var data, leftdata []byte var data, leftdata []byte
for { for {
if data, err = a.read(bufferSize); err != nil { if data, err = a.read(bufferSize); err != nil {
if err == ErrLostConn { if opErr, ok := err.(*net.OpError); ok {
if opErr.Timeout() {
a.worker.err(err)
}
if opErr.Temporary() {
continue
}
break break
} }
a.worker.err(err) a.worker.err(err)
@ -116,16 +120,6 @@ func (a *agent) PreSleep() {
a.write(outpack) a.write(outpack)
} }
func isClosed(err error) bool {
switch {
case err == io.EOF:
fallthrough
case strings.Contains(err.Error(), "use of closed network connection"):
return true
}
return false
}
// read length bytes from the socket // read length bytes from the socket
func (a *agent) read(length int) (data []byte, err error) { func (a *agent) read(length int) (data []byte, err error) {
n := 0 n := 0
@ -133,9 +127,6 @@ func (a *agent) read(length int) (data []byte, err error) {
// read until data can be unpacked // read until data can be unpacked
for i := length; i > 0 || len(data) < minPacketLength; i -= n { for i := length; i > 0 || len(data) < minPacketLength; i -= n {
if n, err = a.rw.Read(buf); err != nil { if n, err = a.rw.Read(buf); err != nil {
if isClosed(err) {
err = ErrLostConn
}
return return
} }
data = append(data, buf[0:n]...) data = append(data, buf[0:n]...)