forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-17 01:13:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

rpc: implement stdio client

Browse source
This commit is contained in:
Martin Holst Swende 2017-12-18 20:32:24 +01:00
parent 4eea6854c5
commit 436bbed89a
No known key found for this signature in database
GPG key ID: 683B438C05A5DDF0
1 changed files with 47 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

53
rpc/client.go
View file

@ -33,6 +33,7 @@ import (
"time"
"github.com/ethereum/go-ethereum/log"
"os"
)
var (
@ -171,6 +172,8 @@ func DialContext(ctx context.Context, rawurl string) (*Client, error) {
return DialHTTP(rawurl)
case "ws", "wss":
return DialWebsocket(ctx, rawurl, "")
case "stdio":
return DialStdIO(ctx)
case "":
return DialIPC(ctx, rawurl)
default:
@ -178,13 +181,51 @@ func DialContext(ctx context.Context, rawurl string) (*Client, error) {
}
}
type StdIOConn struct{}
func (io StdIOConn) Read(b []byte) (n int, err error) {
return os.Stdin.Read(b)
}
func (io StdIOConn) Write(b []byte) (n int, err error) {
return os.Stdout.Write(b)
}
func (io StdIOConn) Close() error {
return nil
}
func (io StdIOConn) LocalAddr() net.Addr {
return &net.UnixAddr{Name: "stdio", Net: "stdio"}
}
func (io StdIOConn) RemoteAddr() net.Addr {
return &net.UnixAddr{Name: "stdio", Net: "stdio"}
}
func (io StdIOConn) SetDeadline(t time.Time) error {
return &net.OpError{Op: "set", Net: "stdio", Source: nil, Addr: nil, Err: errors.New("deadline not supported")}
}
func (io StdIOConn) SetReadDeadline(t time.Time) error {
return &net.OpError{Op: "set", Net: "stdio", Source: nil, Addr: nil, Err: errors.New("deadline not supported")}
}
func (io StdIOConn) SetWriteDeadline(t time.Time) error {
return &net.OpError{Op: "set", Net: "stdio", Source: nil, Addr: nil, Err: errors.New("deadline not supported")}
}
func DialStdIO(ctx context.Context) (*Client, error) {
return newClient(ctx, func(_ context.Context) (net.Conn, error) {
return StdIOConn{}, nil
})
}
func newClient(initctx context.Context, connectFunc func(context.Context) (net.Conn, error)) (*Client, error) {
conn, err := connectFunc(initctx)
if err != nil {
return nil, err
}
_, isHTTP := conn.(*httpConn)
c := &Client{
writeConn: conn,
isHTTP: isHTTP,
@ -524,13 +565,13 @@ func (c *Client) dispatch(conn net.Conn) {
}
case err := <-c.readErr:
log.Debug(fmt.Sprintf("<-readErr: %v", err))
log.Debug("<-readErr", "err", err)
c.closeRequestOps(err)
conn.Close()
reading = false
case newconn := <-c.reconnected:
log.Debug(fmt.Sprintf("<-reconnected: (reading=%t) %v", reading, conn.RemoteAddr()))
log.Debug("<-reconnected", "reading", reading, "remote", conn.RemoteAddr())
if reading {
// Wait for the previous read loop to exit. This is a rare case.
conn.Close()
@ -587,7 +628,7 @@ func (c *Client) closeRequestOps(err error) {
func (c *Client) handleNotification(msg *jsonrpcMessage) {
if !strings.HasSuffix(msg.Method, notificationMethodSuffix) {
log.Debug(fmt.Sprint("dropping non-subscription message: ", msg))
log.Debug("dropping non-subscription message", "msg", msg)
return
}
var subResult struct {
@ -595,7 +636,7 @@ func (c *Client) handleNotification(msg *jsonrpcMessage) {
Result json.RawMessage `json:"result"`
}
if err := json.Unmarshal(msg.Params, &subResult); err != nil {
log.Debug(fmt.Sprint("dropping invalid subscription message: ", msg))
log.Debug("dropping invalid subscription message", "msg", msg)
return
}
if c.subs[subResult.ID] != nil {
@ -606,7 +647,7 @@ func (c *Client) handleNotification(msg *jsonrpcMessage) {
func (c *Client) handleResponse(msg *jsonrpcMessage) {
op := c.respWait[string(msg.ID)]
if op == nil {
log.Debug(fmt.Sprintf("unsolicited response %v", msg))
log.Debug("unsolicited response", "msg", msg)
return
}
delete(c.respWait, string(msg.ID))