You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

163 lines
4.3 KiB

package server
import (
"bufio"
"fmt"
"io"
"net"
"strings"
. "github.com/tendermint/go-common"
"github.com/tendermint/go-wire"
"github.com/tendermint/tmsp/types"
)
// var maxNumberConnections = 2
func StartListener(protoAddr string, app types.Application) (net.Listener, error) {
parts := strings.SplitN(protoAddr, "://", 2)
proto, addr := parts[0], parts[1]
ln, err := net.Listen(proto, addr)
if err != nil {
return nil, err
}
// A goroutine to accept a connection.
go func() {
// semaphore := make(chan struct{}, maxNumberConnections)
for {
// semaphore <- struct{}{}
// Accept a connection
fmt.Println("Waiting for new connection...")
conn, err := ln.Accept()
if err != nil {
Exit("Failed to accept connection")
} else {
fmt.Println("Accepted a new connection")
}
appContext := app.Open()
closeConn := make(chan error, 2) // Push to signal connection closed
responses := make(chan types.Response, 1000) // A channel to buffer responses
// Read requests from conn and deal with them
go handleRequests(appContext, closeConn, conn, responses)
// Pull responses from 'responses' and write them to conn.
go handleResponses(closeConn, responses, conn)
go func() {
// Wait until signal to close connection
errClose := <-closeConn
if errClose != nil {
fmt.Printf("Connection error: %v\n", errClose)
} else {
fmt.Println("Connection was closed.")
}
// Close the connection
err := conn.Close()
if err != nil {
fmt.Printf("Error in closing connection: %v\n", err)
}
// Close the AppContext
err = appContext.Close()
if err != nil {
fmt.Printf("Error in closing app context: %v\n", err)
}
// <-semaphore
}()
}
}()
return ln, nil
}
// Read requests from conn and deal with them
func handleRequests(appC types.AppContext, closeConn chan error, conn net.Conn, responses chan<- types.Response) {
var count int
var bufReader = bufio.NewReader(conn)
for {
var n int
var err error
var req types.Request
wire.ReadBinaryPtrLengthPrefixed(&req, bufReader, 0, &n, &err)
if err != nil {
if err == io.EOF {
closeConn <- fmt.Errorf("Connection closed by client")
} else {
closeConn <- fmt.Errorf("Error in handleRequests: %v", err.Error())
}
return
}
count++
handleRequest(appC, req, responses)
}
}
func handleRequest(appC types.AppContext, req types.Request, responses chan<- types.Response) {
switch req := req.(type) {
case types.RequestEcho:
msg := appC.Echo(req.Message)
responses <- types.ResponseEcho{msg}
case types.RequestFlush:
responses <- types.ResponseFlush{}
case types.RequestInfo:
data := appC.Info()
responses <- types.ResponseInfo{data}
case types.RequestSetOption:
retCode := appC.SetOption(req.Key, req.Value)
responses <- types.ResponseSetOption{retCode}
case types.RequestAppendTx:
events, retCode := appC.AppendTx(req.TxBytes)
responses <- types.ResponseAppendTx{retCode}
for _, event := range events {
responses <- types.ResponseEvent{event}
}
case types.RequestGetHash:
hash, retCode := appC.GetHash()
responses <- types.ResponseGetHash{retCode, hash}
case types.RequestCommit:
retCode := appC.Commit()
responses <- types.ResponseCommit{retCode}
case types.RequestRollback:
retCode := appC.Rollback()
responses <- types.ResponseRollback{retCode}
case types.RequestAddListener:
retCode := appC.AddListener(req.EventKey)
responses <- types.ResponseAddListener{retCode}
case types.RequestRemListener:
retCode := appC.RemListener(req.EventKey)
responses <- types.ResponseRemListener{retCode}
default:
responses <- types.ResponseException{"Unknown request"}
}
}
// Pull responses from 'responses' and write them to conn.
func handleResponses(closeConn chan error, responses <-chan types.Response, conn net.Conn) {
var count int
var bufWriter = bufio.NewWriter(conn)
for {
var res = <-responses
var n int
var err error
wire.WriteBinaryLengthPrefixed(struct{ types.Response }{res}, bufWriter, &n, &err)
if err != nil {
closeConn <- fmt.Errorf("Error in handleResponses: %v", err.Error())
return
}
if _, ok := res.(types.ResponseFlush); ok {
err = bufWriter.Flush()
if err != nil {
closeConn <- fmt.Errorf("Error in handleResponses: %v", err.Error())
return
}
}
count++
}
}