oasis/fastcgi/client.go

package fastcgi

import (
	"bufio"
	"bytes"
	"encoding/binary"
	"fmt"
	"io"
	"log"
	"net/http"
	"net/http/httputil"
	"net/textproto"
	"strconv"
	"strings"
	"sync"
)

type FCGIClient struct {
	mutex     sync.Mutex
	rwc       io.ReadWriteCloser
	h         Header
	buf       bytes.Buffer
	keepAlive bool
	reqId     uint16
}

// Close fcgi connnection
func (client *FCGIClient) Close() {
	client.rwc.Close()
}

func (client *FCGIClient) writeRecord(r *Record) (err error) {
	client.mutex.Lock()
	defer client.mutex.Unlock()
	client.buf.Reset()

	// Write the record to the connection
	b, err := r.toBytes()
	_, err = client.rwc.Write(b)
	return err
}

// We write long data such as FCGI_PARAMS or FCGI_STDIN
// as a stream. If a content's body is larger than
// maxWrite, we will split it into separate records.
// maxWrite is determined by the size of the
// contentLength field in the header (2 bytes)
// Meaning that the max value for contentLength
// we can encode is 65535.
func (c *FCGIClient) writeStream(records []Record) error {
	if len(records) == 0 {
		return nil
	}

	log.Printf("Writing %d records of type %d to stream", len(records), records[0].Header.Type)
	for _, r := range records {
		// Write the piece of the record to the stream
		c.writeRecord(&r)
	}

	// Send an empty record to end the stream.
	// all the records should be of the same
	// type so we'll just use the type from
	// the first item in the slice.
	log.Println("Ending stream")
	end := NewRecord(records[0].Header.Type, nil)
	c.writeRecord(end)
	return nil
}

func readRecord(r io.Reader) (*Record, error) {
	// It's easier to read the header piece of the record
	// into the struct as opposed to doing it piece by
	// piece. But we'll do it this way to be explicit.
	var version uint8
	var recType FCGIRecordType
	var id uint16
	var contentlength uint16
	var paddinglength uint8
	var reserved uint8

	log.Println("Reading header of response record")
	// Let's read the header fields of the record.
	binary.Read(r, binary.BigEndian, &version)
	binary.Read(r, binary.BigEndian, &recType)
	binary.Read(r, binary.BigEndian, &id)
	binary.Read(r, binary.BigEndian, &contentlength)
	binary.Read(r, binary.BigEndian, &paddinglength)
	binary.Read(r, binary.BigEndian, &reserved)
	log.Printf("Ver: %d Type: %d ID: %d Length: %d Padding: %d Reserved; %d", version, recType, id, contentlength, paddinglength, reserved)

	log.Println("Reading record contents")
	readLength := int(contentlength) + int(paddinglength)
	content := make([]byte, readLength)

	if _, err := io.ReadFull(r, content); err != nil {
		return nil, err
	}

	// Remove any padding from the content
	content = content[:contentlength]

	rec := Record{}
	rec.Header.Version = version
	rec.Header.Type = recType
	rec.Header.Id = id
	rec.Header.ContentLength = contentlength
	rec.Header.PaddingLength = paddinglength
	rec.Header.Reserved = reserved
	rec.Content = content

	return &rec, nil
}

func (c *FCGIClient) readResponse() []byte {
	var response bytes.Buffer

	log.Println("-- STARTING STDOUT READ --")
	for {
		r, err := readRecord(c.rwc)

		if err != nil {
			log.Printf("Encountered error when reading the stream: %s", err.Error())
		}

		log.Println("Read a record")

		if r.Header.Type == FCGI_END_REQUEST {
			break
		}

		response.Write(r.Content)
	}
	log.Println("-- END STDOUT READ --")

	return response.Bytes()
}

// func (c *FCGIClient) Get(req *http.Request, fcgiParams map[string]string) *http.Response {
//
// }

func (c *FCGIClient) BeginRequest() error {
	err := c.writeRecord(NewBeginRequestRecord())
	if err != nil {
		return err
	}

	return nil
}

// Do made the request and returns a io.Reader that translates the data read
// from fcgi responder out of fcgi packet before returning it.
func (client *FCGIClient) Do(req *FCGIRequest) (http.Response, error) {
	client.BeginRequest()

	// Write the request context as a stream
	log.Println("Sending FCGI_PARAMS")
	client.writeStream(req.EncodeContext())
	log.Println("Done")

	// body := newWriter(client, FCGI_STDIN)
	// if req != nil {
	// 	io.Copy(body, req)
	// }
	// body.Close()

	// Read the app response from the FCGI_STDOUT stream
	respContent := client.readResponse()

	fmt.Println(parseHttp(respContent))

	return parseHttp(respContent)
}

func parseHttp(raw []byte) (http.Response, error) {
	log.Println("Parsing http")
	bf := bufio.NewReader(bytes.NewReader(raw))
	tp := textproto.NewReader(bf)
	resp := new(http.Response)
	// Parse the first line of the response.
	line, err := tp.ReadLine()
	if err != nil {
		if err == io.EOF {
			err = io.ErrUnexpectedEOF
		}
		return http.Response{}, err
	}
	if i := strings.IndexByte(line, ' '); i == -1 {
		err = &badStringError{"malformed HTTP response", line}
	} else {
		resp.Proto = line[:i]
		resp.Status = strings.TrimLeft(line[i+1:], " ")
	}
	statusCode := resp.Status
	if i := strings.IndexByte(resp.Status, ' '); i != -1 {
		statusCode = resp.Status[:i]
	}
	if len(statusCode) != 3 {
		err = &badStringError{"malformed HTTP status code", statusCode}
	}
	resp.StatusCode, err = strconv.Atoi(statusCode)
	if err != nil || resp.StatusCode < 0 {
		err = &badStringError{"malformed HTTP status code", statusCode}
	}
	var ok bool
	if resp.ProtoMajor, resp.ProtoMinor, ok = http.ParseHTTPVersion(resp.Proto); !ok {
		err = &badStringError{"malformed HTTP version", resp.Proto}
	}
	// Parse the response headers.
	mimeHeader, err := tp.ReadMIMEHeader()
	if err != nil {
		if err == io.EOF {
			err = io.ErrUnexpectedEOF
		}
		return http.Response{}, err
	}
	resp.Header = http.Header(mimeHeader)
	// TODO: fixTransferEncoding ?
	resp.TransferEncoding = resp.Header["Transfer-Encoding"]
	resp.ContentLength, _ = strconv.ParseInt(resp.Header.Get("Content-Length"), 10, 64)

	if chunked(resp.TransferEncoding) {
		resp.Body = io.NopCloser(httputil.NewChunkedReader(bf))
	} else {
		resp.Body = io.NopCloser(bf)
	}

	log.Println("Done")

	return *resp, nil
}