203 lines
6.8 KiB
Go
203 lines
6.8 KiB
Go
package gorqlite
|
|
|
|
/*
|
|
this file has low level stuff:
|
|
|
|
rqliteApiGet()
|
|
rqliteApiPost()
|
|
|
|
There is some code duplication between those and they should
|
|
probably be combined into one function.
|
|
|
|
nothing public here.
|
|
|
|
*/
|
|
|
|
import "bytes"
|
|
import "encoding/json"
|
|
import "errors"
|
|
import "fmt"
|
|
import "io/ioutil"
|
|
import "net/http"
|
|
import "time"
|
|
|
|
/* *****************************************************************
|
|
|
|
method: rqliteApiGet() - for api_STATUS
|
|
|
|
- lowest level interface - does not do any JSON unmarshaling
|
|
- handles retries
|
|
- handles timeouts
|
|
|
|
* *****************************************************************/
|
|
|
|
func (conn *Connection) rqliteApiGet(apiOp apiOperation) ([]byte, error) {
|
|
var responseBody []byte
|
|
trace("%s: rqliteApiGet() called",conn.ID)
|
|
|
|
// only api_STATUS now - maybe someday BACKUP
|
|
if ( apiOp != api_STATUS ) {
|
|
return responseBody, errors.New("rqliteApiGet() called for invalid api operation")
|
|
}
|
|
|
|
// just to be safe, check this
|
|
peersToTry := conn.cluster.makePeerList()
|
|
if ( len(peersToTry) < 1 ) {
|
|
return responseBody, errors.New("I don't have any cluster info")
|
|
}
|
|
trace("%s: I have a peer list %d peers long",conn.ID,len(peersToTry))
|
|
|
|
// failure log is used so that if all peers fail, we can say something
|
|
// about why each failed
|
|
failureLog := make([]string,0)
|
|
|
|
PeerLoop:
|
|
for peerNum, peerToTry := range peersToTry {
|
|
trace("%s: attemping to contact peer %d",conn.ID,peerNum)
|
|
// docs say default GET policy is up to 10 follows automatically
|
|
url := conn.assembleURL(api_STATUS,peerToTry)
|
|
req, err := http.NewRequest("GET",url,nil)
|
|
if ( err != nil ) {
|
|
trace("%s: got error '%s' doing http.NewRequest", conn.ID,err.Error())
|
|
failureLog = append(failureLog,fmt.Sprintf("%s failed due to %s",url,err.Error()))
|
|
continue PeerLoop
|
|
}
|
|
trace("%s: http.NewRequest() OK")
|
|
req.Header.Set("Content-Type","application/json")
|
|
client := &http.Client{}
|
|
client.Timeout = time.Duration(conn.timeout) * time.Second
|
|
response, err := client.Do(req)
|
|
if ( err != nil ) {
|
|
trace("%s: got error '%s' doing client.Do", conn.ID,err.Error())
|
|
failureLog = append(failureLog,fmt.Sprintf("%s failed due to %s",url,err.Error()))
|
|
continue PeerLoop
|
|
}
|
|
defer response.Body.Close()
|
|
trace("%s: client.Do() OK")
|
|
responseBody, err := ioutil.ReadAll(response.Body)
|
|
if ( err != nil ) {
|
|
trace("%s: got error '%s' doing ioutil.ReadAll", conn.ID,err.Error())
|
|
failureLog = append(failureLog,fmt.Sprintf("%s failed due to %s",url,err.Error()))
|
|
continue PeerLoop
|
|
}
|
|
trace("%s: ioutil.ReadAll() OK")
|
|
if ( response.Status != "200 OK" ) {
|
|
trace("%s: got code %s",conn.ID,response.Status)
|
|
failureLog = append(failureLog,fmt.Sprintf("%s failed, got: %s",url,response.Status))
|
|
continue PeerLoop
|
|
}
|
|
// if we got here, we succeeded
|
|
trace("%s: api call OK, returning",conn.ID)
|
|
return responseBody, nil
|
|
}
|
|
|
|
// if we got here, all peers failed. Let's build a verbose error message
|
|
var stringBuffer bytes.Buffer
|
|
stringBuffer.WriteString("tried all peers unsuccessfully. here are the results:\n")
|
|
for n, v := range failureLog {
|
|
stringBuffer.WriteString(fmt.Sprintf(" peer #%d: %s\n",n,v))
|
|
}
|
|
return responseBody, errors.New(stringBuffer.String())
|
|
}
|
|
|
|
/* *****************************************************************
|
|
|
|
method: rqliteApiPost() - for api_QUERY and api_WRITE
|
|
|
|
- lowest level interface - does not do any JSON unmarshaling
|
|
- handles 301s, etc.
|
|
- handles retries
|
|
- handles timeouts
|
|
|
|
it is called with an apiOperation type because the URL it will use varies
|
|
depending on the API operation type (api_QUERY vs. api_WRITE)
|
|
|
|
* *****************************************************************/
|
|
|
|
func (conn *Connection) rqliteApiPost (apiOp apiOperation, sqlStatements []string) ([]byte, error) {
|
|
var responseBody []byte
|
|
|
|
switch (apiOp) {
|
|
case api_QUERY:
|
|
trace("%s: rqliteApiGet() post called for a QUERY of %d statements",conn.ID,len(sqlStatements))
|
|
case api_WRITE:
|
|
trace("%s: rqliteApiGet() post called for a QUERY of %d statements",conn.ID,len(sqlStatements))
|
|
default:
|
|
return responseBody, errors.New("weird! called for an invalid apiOperation in rqliteApiPost()")
|
|
}
|
|
|
|
// jsonify the statements. not really needed in the
|
|
// case of api_STATUS but doesn't hurt
|
|
|
|
jStatements , err := json.Marshal(sqlStatements)
|
|
if ( err != nil ) { return nil, err }
|
|
|
|
// just to be safe, check this
|
|
peersToTry := conn.cluster.makePeerList()
|
|
if ( len(peersToTry) < 1 ) {
|
|
return responseBody, errors.New("I don't have any cluster info")
|
|
}
|
|
|
|
// failure log is used so that if all peers fail, we can say something
|
|
// about why each failed
|
|
failureLog := make([]string,0)
|
|
|
|
PeerLoop:
|
|
for peerNum, peer := range peersToTry {
|
|
trace("%s: trying peer #%d",conn.ID,peerNum)
|
|
|
|
// we're doing a post, and the RFCs say that if you get a 301, it's not
|
|
// automatically followed, so we have to do that ourselves
|
|
|
|
responseStatus := "Haven't Tried Yet"
|
|
var url string
|
|
for ( responseStatus == "Haven't Tried Yet" || responseStatus == "301 Moved Permanently" ) {
|
|
url = conn.assembleURL(apiOp,peer)
|
|
req, err := http.NewRequest("POST",url,bytes.NewBuffer(jStatements))
|
|
if ( err != nil ) {
|
|
trace("%s: got error '%s' doing http.NewRequest", conn.ID,err.Error())
|
|
failureLog = append(failureLog,fmt.Sprintf("%s failed due to %s",url,err.Error()))
|
|
continue PeerLoop
|
|
}
|
|
req.Header.Set("Content-Type","application/json")
|
|
client := &http.Client{}
|
|
response, err := client.Do(req)
|
|
if ( err != nil ) {
|
|
trace("%s: got error '%s' doing client.Do", conn.ID,err.Error())
|
|
failureLog = append(failureLog,fmt.Sprintf("%s failed due to %s",url,err.Error()))
|
|
continue PeerLoop
|
|
}
|
|
defer response.Body.Close()
|
|
responseBody, err = ioutil.ReadAll(response.Body)
|
|
if ( err != nil ) {
|
|
trace("%s: got error '%s' doing ioutil.ReadAll", conn.ID,err.Error())
|
|
failureLog = append(failureLog,fmt.Sprintf("%s failed due to %s",url,err.Error()))
|
|
continue PeerLoop
|
|
}
|
|
responseStatus = response.Status
|
|
if ( responseStatus == "301 Moved Permanently" ) {
|
|
v := response.Header["Location"]
|
|
failureLog = append(failureLog,fmt.Sprintf("%s redirected me to %s",url,v[0]))
|
|
url = v[0]
|
|
continue PeerLoop
|
|
} else if ( responseStatus == "200 OK" ) {
|
|
trace("%s: api call OK, returning",conn.ID)
|
|
return responseBody, nil
|
|
} else {
|
|
trace("%s: got error in responseStatus: %s", conn.ID, responseStatus)
|
|
failureLog = append(failureLog,fmt.Sprintf("%s failed, got: %s",url,response.Status))
|
|
continue PeerLoop
|
|
}
|
|
}
|
|
}
|
|
|
|
// if we got here, all peers failed. Let's build a verbose error message
|
|
var stringBuffer bytes.Buffer
|
|
stringBuffer.WriteString("tried all peers unsuccessfully. here are the results:\n")
|
|
for n, v := range failureLog {
|
|
stringBuffer.WriteString(fmt.Sprintf(" peer #%d: %s\n",n,v))
|
|
}
|
|
return responseBody, errors.New(stringBuffer.String())
|
|
}
|
|
|