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.

238 lines
5.8 KiB

7 years ago
7 years ago
8 years ago
8 years ago
8 years ago
  1. package server
  2. import (
  3. "bufio"
  4. "fmt"
  5. "io"
  6. "net"
  7. "sync"
  8. "github.com/tendermint/tendermint/abci/types"
  9. tmnet "github.com/tendermint/tendermint/libs/net"
  10. "github.com/tendermint/tendermint/libs/service"
  11. )
  12. // var maxNumberConnections = 2
  13. type SocketServer struct {
  14. service.BaseService
  15. proto string
  16. addr string
  17. listener net.Listener
  18. connsMtx sync.Mutex
  19. conns map[int]net.Conn
  20. nextConnID int
  21. appMtx sync.Mutex
  22. app types.Application
  23. }
  24. func NewSocketServer(protoAddr string, app types.Application) service.Service {
  25. proto, addr := tmnet.ProtocolAndAddress(protoAddr)
  26. s := &SocketServer{
  27. proto: proto,
  28. addr: addr,
  29. listener: nil,
  30. app: app,
  31. conns: make(map[int]net.Conn),
  32. }
  33. s.BaseService = *service.NewBaseService(nil, "ABCIServer", s)
  34. return s
  35. }
  36. func (s *SocketServer) OnStart() error {
  37. if err := s.BaseService.OnStart(); err != nil {
  38. return err
  39. }
  40. ln, err := net.Listen(s.proto, s.addr)
  41. if err != nil {
  42. return err
  43. }
  44. s.listener = ln
  45. go s.acceptConnectionsRoutine()
  46. return nil
  47. }
  48. func (s *SocketServer) OnStop() {
  49. s.BaseService.OnStop()
  50. if err := s.listener.Close(); err != nil {
  51. s.Logger.Error("Error closing listener", "err", err)
  52. }
  53. s.connsMtx.Lock()
  54. defer s.connsMtx.Unlock()
  55. for id, conn := range s.conns {
  56. delete(s.conns, id)
  57. if err := conn.Close(); err != nil {
  58. s.Logger.Error("Error closing connection", "id", id, "conn", conn, "err", err)
  59. }
  60. }
  61. }
  62. func (s *SocketServer) addConn(conn net.Conn) int {
  63. s.connsMtx.Lock()
  64. defer s.connsMtx.Unlock()
  65. connID := s.nextConnID
  66. s.nextConnID++
  67. s.conns[connID] = conn
  68. return connID
  69. }
  70. // deletes conn even if close errs
  71. func (s *SocketServer) rmConn(connID int) error {
  72. s.connsMtx.Lock()
  73. defer s.connsMtx.Unlock()
  74. conn, ok := s.conns[connID]
  75. if !ok {
  76. return fmt.Errorf("connection %d does not exist", connID)
  77. }
  78. delete(s.conns, connID)
  79. return conn.Close()
  80. }
  81. func (s *SocketServer) acceptConnectionsRoutine() {
  82. for {
  83. // Accept a connection
  84. s.Logger.Info("Waiting for new connection...")
  85. conn, err := s.listener.Accept()
  86. if err != nil {
  87. if !s.IsRunning() {
  88. return // Ignore error from listener closing.
  89. }
  90. s.Logger.Error("Failed to accept connection: " + err.Error())
  91. continue
  92. }
  93. s.Logger.Info("Accepted a new connection")
  94. connID := s.addConn(conn)
  95. closeConn := make(chan error, 2) // Push to signal connection closed
  96. responses := make(chan *types.Response, 1000) // A channel to buffer responses
  97. // Read requests from conn and deal with them
  98. go s.handleRequests(closeConn, conn, responses)
  99. // Pull responses from 'responses' and write them to conn.
  100. go s.handleResponses(closeConn, conn, responses)
  101. // Wait until signal to close connection
  102. go s.waitForClose(closeConn, connID)
  103. }
  104. }
  105. func (s *SocketServer) waitForClose(closeConn chan error, connID int) {
  106. err := <-closeConn
  107. switch {
  108. case err == io.EOF:
  109. s.Logger.Error("Connection was closed by client")
  110. case err != nil:
  111. s.Logger.Error("Connection error", "error", err)
  112. default:
  113. // never happens
  114. s.Logger.Error("Connection was closed.")
  115. }
  116. // Close the connection
  117. if err := s.rmConn(connID); err != nil {
  118. s.Logger.Error("Error in closing connection", "error", err)
  119. }
  120. }
  121. // Read requests from conn and deal with them
  122. func (s *SocketServer) handleRequests(closeConn chan error, conn io.Reader, responses chan<- *types.Response) {
  123. var count int
  124. var bufReader = bufio.NewReader(conn)
  125. defer func() {
  126. // make sure to recover from any app-related panics to allow proper socket cleanup
  127. r := recover()
  128. if r != nil {
  129. closeConn <- fmt.Errorf("recovered from panic: %v", r)
  130. s.appMtx.Unlock()
  131. }
  132. }()
  133. for {
  134. var req = &types.Request{}
  135. err := types.ReadMessage(bufReader, req)
  136. if err != nil {
  137. if err == io.EOF {
  138. closeConn <- err
  139. } else {
  140. closeConn <- fmt.Errorf("error reading message: %v", err)
  141. }
  142. return
  143. }
  144. s.appMtx.Lock()
  145. count++
  146. s.handleRequest(req, responses)
  147. s.appMtx.Unlock()
  148. }
  149. }
  150. func (s *SocketServer) handleRequest(req *types.Request, responses chan<- *types.Response) {
  151. switch r := req.Value.(type) {
  152. case *types.Request_Echo:
  153. responses <- types.ToResponseEcho(r.Echo.Message)
  154. case *types.Request_Flush:
  155. responses <- types.ToResponseFlush()
  156. case *types.Request_Info:
  157. res := s.app.Info(*r.Info)
  158. responses <- types.ToResponseInfo(res)
  159. case *types.Request_SetOption:
  160. res := s.app.SetOption(*r.SetOption)
  161. responses <- types.ToResponseSetOption(res)
  162. case *types.Request_DeliverTx:
  163. res := s.app.DeliverTx(*r.DeliverTx)
  164. responses <- types.ToResponseDeliverTx(res)
  165. case *types.Request_CheckTx:
  166. res := s.app.CheckTx(*r.CheckTx)
  167. responses <- types.ToResponseCheckTx(res)
  168. case *types.Request_Commit:
  169. res := s.app.Commit()
  170. responses <- types.ToResponseCommit(res)
  171. case *types.Request_Query:
  172. res := s.app.Query(*r.Query)
  173. responses <- types.ToResponseQuery(res)
  174. case *types.Request_InitChain:
  175. res := s.app.InitChain(*r.InitChain)
  176. responses <- types.ToResponseInitChain(res)
  177. case *types.Request_BeginBlock:
  178. res := s.app.BeginBlock(*r.BeginBlock)
  179. responses <- types.ToResponseBeginBlock(res)
  180. case *types.Request_EndBlock:
  181. res := s.app.EndBlock(*r.EndBlock)
  182. responses <- types.ToResponseEndBlock(res)
  183. default:
  184. responses <- types.ToResponseException("Unknown request")
  185. }
  186. }
  187. // Pull responses from 'responses' and write them to conn.
  188. func (s *SocketServer) handleResponses(closeConn chan error, conn io.Writer, responses <-chan *types.Response) {
  189. var count int
  190. var bufWriter = bufio.NewWriter(conn)
  191. for {
  192. var res = <-responses
  193. err := types.WriteMessage(res, bufWriter)
  194. if err != nil {
  195. closeConn <- fmt.Errorf("error writing message: %v", err.Error())
  196. return
  197. }
  198. if _, ok := res.Value.(*types.Response_Flush); ok {
  199. err = bufWriter.Flush()
  200. if err != nil {
  201. closeConn <- fmt.Errorf("error flushing write buffer: %v", err.Error())
  202. return
  203. }
  204. }
  205. count++
  206. }
  207. }