2
0

Separate receiving and processing a message

This commit is contained in:
Jack Christensen
2013-04-06 16:52:06 -04:00
parent 0cbf3b6f9e
commit 577f9707f7
+28 -18
View File
@@ -38,10 +38,10 @@ func Connect(options map[string]string) (c *conn, err error) {
c.txStartupMessage(msg) c.txStartupMessage(msg)
var response interface{} var response interface{}
response, err = c.rxMsg() response, err = c.processMsg()
for { for {
response, err = c.rxMsg() response, err = c.processMsg()
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -69,7 +69,7 @@ func (c *conn) Query(sql string) (rows []map[string]string, err error) {
var response interface{} var response interface{}
for { for {
response, err = c.rxMsg() response, err = c.processMsg()
if err != nil { if err != nil {
fmt.Println(err) fmt.Println(err)
return nil, err return nil, err
@@ -95,34 +95,33 @@ func (c *conn) sendSimpleQuery(sql string) (err error) {
return err return err
} }
func (c *conn) rxMsg() (msg interface{}, err error) { func (c *conn) processMsg() (msg interface{}, err error) {
var t byte var t byte
var bodySize int32 var body []byte
t, bodySize, err = c.rxMsgHeader() t, body, err = c.rxMsg()
if err != nil { if err != nil {
return nil, err return
} }
var buf []byte return c.parseMsg(t, body)
if buf, err = c.rxMsgBody(bodySize); err != nil { }
return nil, err
}
func (c *conn) parseMsg(t byte, body []byte) (msg interface{}, err error) {
switch t { switch t {
case 'K': case 'K':
return c.rxBackendKeyData(buf), nil return c.rxBackendKeyData(body), nil
case 'R': case 'R':
return c.rxAuthenticationX(buf) return c.rxAuthenticationX(body)
case 'S': case 'S':
return c.rxParameterStatus(buf) return c.rxParameterStatus(body)
case 'Z': case 'Z':
return c.rxReadyForQuery(buf), nil return c.rxReadyForQuery(body), nil
case 'T': case 'T':
return c.rxRowDescription(buf) return c.rxRowDescription(body)
case 'D': case 'D':
return c.rxDataRow(buf) return c.rxDataRow(body)
case 'C': case 'C':
return c.rxCommandComplete(buf), nil return c.rxCommandComplete(body), nil
default: default:
return nil, fmt.Errorf("Received unknown message type: %c", t) return nil, fmt.Errorf("Received unknown message type: %c", t)
} }
@@ -130,6 +129,17 @@ func (c *conn) rxMsg() (msg interface{}, err error) {
panic("Unreachable") panic("Unreachable")
} }
func (c *conn) rxMsg() (t byte, body []byte, err error) {
var bodySize int32
t, bodySize, err = c.rxMsgHeader()
if err != nil {
return
}
body, err = c.rxMsgBody(bodySize)
return
}
func (c *conn) rxMsgHeader() (t byte, bodySize int32, err error) { func (c *conn) rxMsgHeader() (t byte, bodySize int32, err error) {
buf := c.buf[:5] buf := c.buf[:5]
if _, err = io.ReadFull(c.conn, buf); err != nil { if _, err = io.ReadFull(c.conn, buf); err != nil {