aboutsummaryrefslogtreecommitdiffstats
path: root/rpc
diff options
context:
space:
mode:
Diffstat (limited to 'rpc')
-rw-r--r--rpc/api/utils.go41
-rw-r--r--rpc/comms/comms.go7
-rw-r--r--rpc/comms/ipc.go37
-rw-r--r--rpc/comms/ipc_unix.go77
-rw-r--r--rpc/comms/ipc_windows.go699
-rw-r--r--rpc/shared/types.go27
6 files changed, 887 insertions, 1 deletions
diff --git a/rpc/api/utils.go b/rpc/api/utils.go
index 76f00c251..a62058140 100644
--- a/rpc/api/utils.go
+++ b/rpc/api/utils.go
@@ -8,6 +8,7 @@ import (
"github.com/ethereum/go-ethereum/eth"
"github.com/ethereum/go-ethereum/rpc/codec"
"github.com/ethereum/go-ethereum/xeth"
+ "github.com/ethereum/go-ethereum/rpc/shared"
)
const (
@@ -34,3 +35,43 @@ func ParseApiString(apistr string, codec codec.Codec, xeth *xeth.XEth, eth *eth.
return apis, nil
}
+
+// combines multiple API's
+type mergedApi struct {
+ apis map[string]EthereumApi
+}
+
+// create new merged api instance
+func newMergedApi(apis ...EthereumApi) *mergedApi {
+ mergedApi := new(mergedApi)
+ mergedApi.apis = make(map[string]EthereumApi)
+
+ for _, api := range apis {
+ for _, method := range api.Methods() {
+ mergedApi.apis[method] = api
+ }
+ }
+ return mergedApi
+}
+
+// Supported RPC methods
+func (self *mergedApi) Methods() []string {
+ all := make([]string, len(self.apis))
+ for method, _ := range self.apis {
+ all = append(all, method)
+ }
+ return all
+}
+
+// Call the correct API's Execute method for the given request
+func (self *mergedApi) Execute(req *shared.Request) (interface{}, error) {
+ if api, found := self.apis[req.Method]; found {
+ return api.Execute(req)
+ }
+ return nil, shared.NewNotImplementedError(req.Method)
+}
+
+// Merge multiple API's to a single API instance
+func Merge(apis ...EthereumApi) EthereumApi {
+ return newMergedApi(apis...)
+}
diff --git a/rpc/comms/comms.go b/rpc/comms/comms.go
new file mode 100644
index 000000000..244f5a7a6
--- /dev/null
+++ b/rpc/comms/comms.go
@@ -0,0 +1,7 @@
+package comms
+
+type EthereumClient interface {
+ Close()
+ Send(interface{}) error
+ Recv() (interface{}, error)
+}
diff --git a/rpc/comms/ipc.go b/rpc/comms/ipc.go
new file mode 100644
index 000000000..a75039d17
--- /dev/null
+++ b/rpc/comms/ipc.go
@@ -0,0 +1,37 @@
+package comms
+
+import (
+ "github.com/ethereum/go-ethereum/rpc/api"
+ "github.com/ethereum/go-ethereum/rpc/codec"
+)
+
+type IpcConfig struct {
+ Endpoint string
+}
+
+type ipcClient struct {
+ c codec.ApiCoder
+}
+
+func (self *ipcClient) Close() {
+ self.c.Close()
+}
+
+func (self *ipcClient) Send(req interface{}) error {
+ return self.c.WriteResponse(req)
+}
+
+func (self *ipcClient) Recv() (interface{}, error) {
+ return self.c.ReadResponse()
+}
+
+// Create a new IPC client, UNIX domain socket on posix, named pipe on Windows
+func NewIpcClient(cfg IpcConfig, codec codec.Codec) (*ipcClient, error) {
+ return newIpcClient(cfg, codec)
+}
+
+// Start IPC server
+func StartIpc(cfg IpcConfig, codec codec.Codec, apis ...api.EthereumApi) error {
+ offeredApi := api.Merge(apis...)
+ return startIpc(cfg, codec, offeredApi)
+}
diff --git a/rpc/comms/ipc_unix.go b/rpc/comms/ipc_unix.go
new file mode 100644
index 000000000..bb09d9547
--- /dev/null
+++ b/rpc/comms/ipc_unix.go
@@ -0,0 +1,77 @@
+// +build darwin dragonfly freebsd linux nacl netbsd openbsd solaris
+
+package comms
+
+import (
+ "io"
+ "net"
+ "os"
+
+ "github.com/ethereum/go-ethereum/logger"
+ "github.com/ethereum/go-ethereum/logger/glog"
+ "github.com/ethereum/go-ethereum/rpc/api"
+ "github.com/ethereum/go-ethereum/rpc/codec"
+ "github.com/ethereum/go-ethereum/rpc/shared"
+)
+
+func newIpcClient(cfg IpcConfig, codec codec.Codec) (*ipcClient, error) {
+ c, err := net.DialUnix("unix", nil, &net.UnixAddr{cfg.Endpoint, "unix"})
+ if err != nil {
+ return nil, err
+ }
+
+ return &ipcClient{codec.New(c)}, nil
+}
+
+func startIpc(cfg IpcConfig, codec codec.Codec, api api.EthereumApi) error {
+ os.Remove(cfg.Endpoint) // in case it still exists from a previous run
+
+ l, err := net.ListenUnix("unix", &net.UnixAddr{Name: cfg.Endpoint, Net: "unix"})
+ if err != nil {
+ return err
+ }
+ os.Chmod(cfg.Endpoint, 0600)
+
+ go func() {
+ for {
+ conn, err := l.AcceptUnix()
+ if err != nil {
+ glog.V(logger.Error).Infof("Error accepting ipc connection - %v\n", err)
+ continue
+ }
+
+ go func(conn net.Conn) {
+ codec := codec.New(conn)
+
+ for {
+ req, err := codec.ReadRequest()
+ if err == io.EOF {
+ codec.Close()
+ return
+ } else if err != nil {
+ glog.V(logger.Error).Infof("IPC recv err - %v\n", err)
+ codec.Close()
+ return
+ }
+
+ var rpcResponse interface{}
+ res, err := api.Execute(req)
+
+ rpcResponse = shared.NewRpcResponse(req.Id, req.Jsonrpc, res, err)
+ err = codec.WriteResponse(rpcResponse)
+ if err != nil {
+ glog.V(logger.Error).Infof("IPC send err - %v\n", err)
+ codec.Close()
+ return
+ }
+ }
+ }(conn)
+ }
+
+ os.Remove(cfg.Endpoint)
+ }()
+
+ glog.V(logger.Info).Infof("IPC service started (%s)\n", cfg.Endpoint)
+
+ return nil
+}
diff --git a/rpc/comms/ipc_windows.go b/rpc/comms/ipc_windows.go
new file mode 100644
index 000000000..ff9015d03
--- /dev/null
+++ b/rpc/comms/ipc_windows.go
@@ -0,0 +1,699 @@
+// +build windows
+
+package comms
+
+import (
+ "fmt"
+ "io"
+ "net"
+ "os"
+ "sync"
+ "syscall"
+ "time"
+ "unsafe"
+
+ "github.com/ethereum/go-ethereum/logger"
+ "github.com/ethereum/go-ethereum/logger/glog"
+ "github.com/ethereum/go-ethereum/rpc/api"
+ "github.com/ethereum/go-ethereum/rpc/codec"
+ "github.com/ethereum/go-ethereum/rpc/shared"
+)
+
+var (
+ modkernel32 = syscall.NewLazyDLL("kernel32.dll")
+
+ procCreateNamedPipeW = modkernel32.NewProc("CreateNamedPipeW")
+ procConnectNamedPipe = modkernel32.NewProc("ConnectNamedPipe")
+ procDisconnectNamedPipe = modkernel32.NewProc("DisconnectNamedPipe")
+ procWaitNamedPipeW = modkernel32.NewProc("WaitNamedPipeW")
+ procCreateEventW = modkernel32.NewProc("CreateEventW")
+ procGetOverlappedResult = modkernel32.NewProc("GetOverlappedResult")
+ procCancelIoEx = modkernel32.NewProc("CancelIoEx")
+)
+
+func createNamedPipe(name *uint16, openMode uint32, pipeMode uint32, maxInstances uint32, outBufSize uint32, inBufSize uint32, defaultTimeout uint32, sa *syscall.SecurityAttributes) (handle syscall.Handle, err error) {
+ r0, _, e1 := syscall.Syscall9(procCreateNamedPipeW.Addr(), 8, uintptr(unsafe.Pointer(name)), uintptr(openMode), uintptr(pipeMode), uintptr(maxInstances), uintptr(outBufSize), uintptr(inBufSize), uintptr(defaultTimeout), uintptr(unsafe.Pointer(sa)), 0)
+ handle = syscall.Handle(r0)
+ if handle == syscall.InvalidHandle {
+ if e1 != 0 {
+ err = error(e1)
+ } else {
+ err = syscall.EINVAL
+ }
+ }
+ return
+}
+
+func cancelIoEx(handle syscall.Handle, overlapped *syscall.Overlapped) (err error) {
+ r1, _, e1 := syscall.Syscall(procCancelIoEx.Addr(), 2, uintptr(handle), uintptr(unsafe.Pointer(overlapped)), 0)
+ if r1 == 0 {
+ if e1 != 0 {
+ err = error(e1)
+ } else {
+ err = syscall.EINVAL
+ }
+ }
+ return
+}
+
+func connectNamedPipe(handle syscall.Handle, overlapped *syscall.Overlapped) (err error) {
+ r1, _, e1 := syscall.Syscall(procConnectNamedPipe.Addr(), 2, uintptr(handle), uintptr(unsafe.Pointer(overlapped)), 0)
+ if r1 == 0 {
+ if e1 != 0 {
+ err = error(e1)
+ } else {
+ err = syscall.EINVAL
+ }
+ }
+ return
+}
+
+func disconnectNamedPipe(handle syscall.Handle) (err error) {
+ r1, _, e1 := syscall.Syscall(procDisconnectNamedPipe.Addr(), 1, uintptr(handle), 0, 0)
+ if r1 == 0 {
+ if e1 != 0 {
+ err = error(e1)
+ } else {
+ err = syscall.EINVAL
+ }
+ }
+ return
+}
+
+func waitNamedPipe(name *uint16, timeout uint32) (err error) {
+ r1, _, e1 := syscall.Syscall(procWaitNamedPipeW.Addr(), 2, uintptr(unsafe.Pointer(name)), uintptr(timeout), 0)
+ if r1 == 0 {
+ if e1 != 0 {
+ err = error(e1)
+ } else {
+ err = syscall.EINVAL
+ }
+ }
+ return
+}
+
+func createEvent(sa *syscall.SecurityAttributes, manualReset bool, initialState bool, name *uint16) (handle syscall.Handle, err error) {
+ var _p0 uint32
+ if manualReset {
+ _p0 = 1
+ } else {
+ _p0 = 0
+ }
+ var _p1 uint32
+ if initialState {
+ _p1 = 1
+ } else {
+ _p1 = 0
+ }
+ r0, _, e1 := syscall.Syscall6(procCreateEventW.Addr(), 4, uintptr(unsafe.Pointer(sa)), uintptr(_p0), uintptr(_p1), uintptr(unsafe.Pointer(name)), 0, 0)
+ handle = syscall.Handle(r0)
+ if handle == syscall.InvalidHandle {
+ if e1 != 0 {
+ err = error(e1)
+ } else {
+ err = syscall.EINVAL
+ }
+ }
+ return
+}
+
+func getOverlappedResult(handle syscall.Handle, overlapped *syscall.Overlapped, transferred *uint32, wait bool) (err error) {
+ var _p0 uint32
+ if wait {
+ _p0 = 1
+ } else {
+ _p0 = 0
+ }
+ r1, _, e1 := syscall.Syscall6(procGetOverlappedResult.Addr(), 4, uintptr(handle), uintptr(unsafe.Pointer(overlapped)), uintptr(unsafe.Pointer(transferred)), uintptr(_p0), 0, 0)
+ if r1 == 0 {
+ if e1 != 0 {
+ err = error(e1)
+ } else {
+ err = syscall.EINVAL
+ }
+ }
+ return
+}
+
+
+const (
+ // openMode
+ pipe_access_duplex = 0x3
+ pipe_access_inbound = 0x1
+ pipe_access_outbound = 0x2
+
+ // openMode write flags
+ file_flag_first_pipe_instance = 0x00080000
+ file_flag_write_through = 0x80000000
+ file_flag_overlapped = 0x40000000
+
+ // openMode ACL flags
+ write_dac = 0x00040000
+ write_owner = 0x00080000
+ access_system_security = 0x01000000
+
+ // pipeMode
+ pipe_type_byte = 0x0
+ pipe_type_message = 0x4
+
+ // pipeMode read mode flags
+ pipe_readmode_byte = 0x0
+ pipe_readmode_message = 0x2
+
+ // pipeMode wait mode flags
+ pipe_wait = 0x0
+ pipe_nowait = 0x1
+
+ // pipeMode remote-client mode flags
+ pipe_accept_remote_clients = 0x0
+ pipe_reject_remote_clients = 0x8
+
+ pipe_unlimited_instances = 255
+
+ nmpwait_wait_forever = 0xFFFFFFFF
+
+ // the two not-an-errors below occur if a client connects to the pipe between
+ // the server's CreateNamedPipe and ConnectNamedPipe calls.
+ error_no_data syscall.Errno = 0xE8
+ error_pipe_connected syscall.Errno = 0x217
+ error_pipe_busy syscall.Errno = 0xE7
+ error_sem_timeout syscall.Errno = 0x79
+
+ error_bad_pathname syscall.Errno = 0xA1
+ error_invalid_name syscall.Errno = 0x7B
+
+ error_io_incomplete syscall.Errno = 0x3e4
+)
+
+var _ net.Conn = (*PipeConn)(nil)
+var _ net.Listener = (*PipeListener)(nil)
+
+// ErrClosed is the error returned by PipeListener.Accept when Close is called
+// on the PipeListener.
+var ErrClosed = PipeError{"Pipe has been closed.", false}
+
+// PipeError is an error related to a call to a pipe
+type PipeError struct {
+ msg string
+ timeout bool
+}
+
+// Error implements the error interface
+func (e PipeError) Error() string {
+ return e.msg
+}
+
+// Timeout implements net.AddrError.Timeout()
+func (e PipeError) Timeout() bool {
+ return e.timeout
+}
+
+// Temporary implements net.AddrError.Temporary()
+func (e PipeError) Temporary() bool {
+ return false
+}
+
+// Dial connects to a named pipe with the given address. If the specified pipe is not available,
+// it will wait indefinitely for the pipe to become available.
+//
+// The address must be of the form \\.\\pipe\<name> for local pipes and \\<computer>\pipe\<name>
+// for remote pipes.
+//
+// Dial will return a PipeError if you pass in a badly formatted pipe name.
+//
+// Examples:
+// // local pipe
+// conn, err := Dial(`\\.\pipe\mypipename`)
+//
+// // remote pipe
+// conn, err := Dial(`\\othercomp\pipe\mypipename`)
+func Dial(address string) (*PipeConn, error) {
+ for {
+ conn, err := dial(address, nmpwait_wait_forever)
+ if err == nil {
+ return conn, nil
+ }
+ if isPipeNotReady(err) {
+ <-time.After(100 * time.Millisecond)
+ continue
+ }
+ return nil, err
+ }
+}
+
+// DialTimeout acts like Dial, but will time out after the duration of timeout
+func DialTimeout(address string, timeout time.Duration) (*PipeConn, error) {
+ deadline := time.Now().Add(timeout)
+
+ now := time.Now()
+ for now.Before(deadline) {
+ millis := uint32(deadline.Sub(now) / time.Millisecond)
+ conn, err := dial(address, millis)
+ if err == nil {
+ return conn, nil
+ }
+ if err == error_sem_timeout {
+ // This is WaitNamedPipe's timeout error, so we know we're done
+ return nil, PipeError{fmt.Sprintf(
+ "Timed out waiting for pipe '%s' to come available", address), true}
+ }
+ if isPipeNotReady(err) {
+ left := deadline.Sub(time.Now())
+ retry := 100 * time.Millisecond
+ if left > retry {
+ <-time.After(retry)
+ } else {
+ <-time.After(left - time.Millisecond)
+ }
+ now = time.Now()
+ continue
+ }
+ return nil, err
+ }
+ return nil, PipeError{fmt.Sprintf(
+ "Timed out waiting for pipe '%s' to come available", address), true}
+}
+
+// isPipeNotReady checks the error to see if it indicates the pipe is not ready
+func isPipeNotReady(err error) bool {
+ // Pipe Busy means another client just grabbed the open pipe end,
+ // and the server hasn't made a new one yet.
+ // File Not Found means the server hasn't created the pipe yet.
+ // Neither is a fatal error.
+
+ return err == syscall.ERROR_FILE_NOT_FOUND || err == error_pipe_busy
+}
+
+// newOverlapped creates a structure used to track asynchronous
+// I/O requests that have been issued.
+func newOverlapped() (*syscall.Overlapped, error) {
+ event, err := createEvent(nil, true, true, nil)
+ if err != nil {
+ return nil, err
+ }
+ return &syscall.Overlapped{HEvent: event}, nil
+}
+
+// waitForCompletion waits for an asynchronous I/O request referred to by overlapped to complete.
+// This function returns the number of bytes transferred by the operation and an error code if
+// applicable (nil otherwise).
+func waitForCompletion(handle syscall.Handle, overlapped *syscall.Overlapped) (uint32, error) {
+ _, err := syscall.WaitForSingleObject(overlapped.HEvent, syscall.INFINITE)
+ if err != nil {
+ return 0, err
+ }
+ var transferred uint32
+ err = getOverlappedResult(handle, overlapped, &transferred, true)
+ return transferred, err
+}
+
+// dial is a helper to initiate a connection to a named pipe that has been started by a server.
+// The timeout is only enforced if the pipe server has already created the pipe, otherwise
+// this function will return immediately.
+func dial(address string, timeout uint32) (*PipeConn, error) {
+ name, err := syscall.UTF16PtrFromString(string(address))
+ if err != nil {
+ return nil, err
+ }
+ // If at least one instance of the pipe has been created, this function
+ // will wait timeout milliseconds for it to become available.
+ // It will return immediately regardless of timeout, if no instances
+ // of the named pipe have been created yet.
+ // If this returns with no error, there is a pipe available.
+ if err := waitNamedPipe(name, timeout); err != nil {
+ if err == error_bad_pathname {
+ // badly formatted pipe name
+ return nil, badAddr(address)
+ }
+ return nil, err
+ }
+ pathp, err := syscall.UTF16PtrFromString(address)
+ if err != nil {
+ return nil, err
+ }
+ handle, err := syscall.CreateFile(pathp, syscall.GENERIC_READ|syscall.GENERIC_WRITE,
+ uint32(syscall.FILE_SHARE_READ|syscall.FILE_SHARE_WRITE), nil, syscall.OPEN_EXISTING,
+ syscall.FILE_FLAG_OVERLAPPED, 0)
+ if err != nil {
+ return nil, err
+ }
+ return &PipeConn{handle: handle, addr: PipeAddr(address)}, nil
+}
+
+// Listen returns a new PipeListener that will listen on a pipe with the given
+// address. The address must be of the form \\.\pipe\<name>
+//
+// Listen will return a PipeError for an incorrectly formatted pipe name.
+func Listen(address string) (*PipeListener, error) {
+ handle, err := createPipe(address, true)
+ if err == error_invalid_name {
+ return nil, badAddr(address)
+ }
+ if err != nil {
+ return nil, err
+ }
+ return &PipeListener{
+ addr: PipeAddr(address),
+ handle: handle,
+ }, nil
+}
+
+// PipeListener is a named pipe listener. Clients should typically
+// use variables of type net.Listener instead of assuming named pipe.
+type PipeListener struct {
+ addr PipeAddr
+ handle syscall.Handle
+ closed bool
+
+ // acceptHandle contains the current handle waiting for
+ // an incoming connection or nil.
+ acceptHandle syscall.Handle
+ // acceptOverlapped is set before waiting on a connection.
+ // If not waiting, it is nil.
+ acceptOverlapped *syscall.Overlapped
+ // acceptMutex protects the handle and overlapped structure.
+ acceptMutex sync.Mutex
+}
+
+// Accept implements the Accept method in the net.Listener interface; it
+// waits for the next call and returns a generic net.Conn.
+func (l *PipeListener) Accept() (net.Conn, error) {
+ c, err := l.AcceptPipe()
+ for err == error_no_data {
+ // Ignore clients that connect and immediately disconnect.
+ c, err = l.AcceptPipe()
+ }
+ if err != nil {
+ return nil, err
+ }
+ return c, nil
+}
+
+// AcceptPipe accepts the next incoming call and returns the new connection.
+// It might return an error if a client connected and immediately cancelled
+// the connection.
+func (l *PipeListener) AcceptPipe() (*PipeConn, error) {
+ if l == nil || l.addr == "" || l.closed {
+ return nil, syscall.EINVAL
+ }
+
+ // the first time we call accept, the handle will have been created by the Listen
+ // call. This is to prevent race conditions where the client thinks the server
+ // isn't listening because it hasn't actually called create yet. After the first time, we'll
+ // have to create a new handle each time
+ handle := l.handle
+ if handle == 0 {
+ var err error
+ handle, err = createPipe(string(l.addr), false)
+ if err != nil {
+ return nil, err
+ }
+ } else {
+ l.handle = 0
+ }
+
+ overlapped, err := newOverlapped()
+ if err != nil {
+ return nil, err
+ }
+ defer syscall.CloseHandle(overlapped.HEvent)
+ if err := connectNamedPipe(handle, overlapped); err != nil && err != error_pipe_connected {
+ if err == error_io_incomplete || err == syscall.ERROR_IO_PENDING {
+ l.acceptMutex.Lock()
+ l.acceptOverlapped = overlapped
+ l.acceptHandle = handle
+ l.acceptMutex.Unlock()
+ defer func() {
+ l.acceptMutex.Lock()
+ l.acceptOverlapped = nil
+ l.acceptHandle = 0
+ l.acceptMutex.Unlock()
+ }()
+
+ _, err = waitForCompletion(handle, overlapped)
+ }
+ if err == syscall.ERROR_OPERATION_ABORTED {
+ // Return error compatible to net.Listener.Accept() in case the
+ // listener was closed.
+ return nil, ErrClosed
+ }
+ if err != nil {
+ return nil, err
+ }
+ }
+ return &PipeConn{handle: handle, addr: l.addr}, nil
+}
+
+// Close stops listening on the address.
+// Already Accepted connections are not closed.
+func (l *PipeListener) Close() error {
+ if l.closed {
+ return nil
+ }
+ l.closed = true
+ if l.handle != 0 {
+ err := disconnectNamedPipe(l.handle)
+ if err != nil {
+ return err
+ }
+ err = syscall.CloseHandle(l.handle)
+ if err != nil {
+ return err
+ }
+ l.handle = 0
+ }
+ l.acceptMutex.Lock()
+ defer l.acceptMutex.Unlock()
+ if l.acceptOverlapped != nil && l.acceptHandle != 0 {
+ // Cancel the pending IO. This call does not block, so it is safe
+ // to hold onto the mutex above.
+ if err := cancelIoEx(l.acceptHandle, l.acceptOverlapped); err != nil {
+ return err
+ }
+ err := syscall.CloseHandle(l.acceptOverlapped.HEvent)
+ if err != nil {
+ return err
+ }
+ l.acceptOverlapped.HEvent = 0
+ err = syscall.CloseHandle(l.acceptHandle)
+ if err != nil {
+ return err
+ }
+ l.acceptHandle = 0
+ }
+ return nil
+}
+
+// Addr returns the listener's network address, a PipeAddr.
+func (l *PipeListener) Addr() net.Addr { return l.addr }
+
+// PipeConn is the implementation of the net.Conn interface for named pipe connections.
+type PipeConn struct {
+ handle syscall.Handle
+ addr PipeAddr
+
+ // these aren't actually used yet
+ readDeadline *time.Time
+ writeDeadline *time.Time
+}
+
+type iodata struct {
+ n uint32
+ err error
+}
+
+// completeRequest looks at iodata to see if a request is pending. If so, it waits for it to either complete or to
+// abort due to hitting the specified deadline. Deadline may be set to nil to wait forever. If no request is pending,
+// the content of iodata is returned.
+func (c *PipeConn) completeRequest(data iodata, deadline *time.Time, overlapped *syscall.Overlapped) (int, error) {
+ if data.err == error_io_incomplete || data.err == syscall.ERROR_IO_PENDING {
+ var timer <-chan time.Time
+ if deadline != nil {
+ if timeDiff := deadline.Sub(time.Now()); timeDiff > 0 {
+ timer = time.After(timeDiff)
+ }
+ }
+ done := make(chan iodata)
+ go func() {
+ n, err := waitForCompletion(c.handle, overlapped)
+ done <- iodata{n, err}
+ }()
+ select {
+ case data = <-done:
+ case <-timer:
+ syscall.CancelIoEx(c.handle, overlapped)
+ data = iodata{0, timeout(c.addr.String())}
+ }
+ }
+ // Windows will produce ERROR_BROKEN_PIPE upon closing
+ // a handle on the other end of a connection. Go RPC
+ // expects an io.EOF error in this case.
+ if data.err == syscall.ERROR_BROKEN_PIPE {
+ data.err = io.EOF
+ }
+ return int(data.n), data.err
+}
+
+// Read implements the net.Conn Read method.
+func (c *PipeConn) Read(b []byte) (int, error) {
+ // Use ReadFile() rather than Read() because the latter
+ // contains a workaround that eats ERROR_BROKEN_PIPE.
+ overlapped, err := newOverlapped()
+ if err != nil {
+ return 0, err
+ }
+ defer syscall.CloseHandle(overlapped.HEvent)
+ var n uint32
+ err = syscall.ReadFile(c.handle, b, &n, overlapped)
+ return c.completeRequest(iodata{n, err}, c.readDeadline, overlapped)
+}
+
+// Write implements the net.Conn Write method.
+func (c *PipeConn) Write(b []byte) (int, error) {
+ overlapped, err := newOverlapped()
+ if err != nil {
+ return 0, err
+ }
+ defer syscall.CloseHandle(overlapped.HEvent)
+ var n uint32
+ err = syscall.WriteFile(c.handle, b, &n, overlapped)
+ return c.completeRequest(iodata{n, err}, c.writeDeadline, overlapped)
+}
+
+// Close closes the connection.
+func (c *PipeConn) Close() error {
+ return syscall.CloseHandle(c.handle)
+}
+
+// LocalAddr returns the local network address.
+func (c *PipeConn) LocalAddr() net.Addr {
+ return c.addr
+}
+
+// RemoteAddr returns the remote network address.
+func (c *PipeConn) RemoteAddr() net.Addr {
+ // not sure what to do here, we don't have remote addr....
+ return c.addr
+}
+
+// SetDeadline implements the net.Conn SetDeadline method.
+// Note that timeouts are only supported on Windows Vista/Server 2008 and above
+func (c *PipeConn) SetDeadline(t time.Time) error {
+ c.SetReadDeadline(t)
+ c.SetWriteDeadline(t)
+ return nil
+}
+
+// SetReadDeadline implements the net.Conn SetReadDeadline method.
+// Note that timeouts are only supported on Windows Vista/Server 2008 and above
+func (c *PipeConn) SetReadDeadline(t time.Time) error {
+ c.readDeadline = &t
+ return nil
+}
+
+// SetWriteDeadline implements the net.Conn SetWriteDeadline method.
+// Note that timeouts are only supported on Windows Vista/Server 2008 and above
+func (c *PipeConn) SetWriteDeadline(t time.Time) error {
+ c.writeDeadline = &t
+ return nil
+}
+
+// PipeAddr represents the address of a named pipe.
+type PipeAddr string
+
+// Network returns the address's network name, "pipe".
+func (a PipeAddr) Network() string { return "pipe" }
+
+// String returns the address of the pipe
+func (a PipeAddr) String() string {
+ return string(a)
+}
+
+// createPipe is a helper function to make sure we always create pipes
+// with the same arguments, since subsequent calls to create pipe need
+// to use the same arguments as the first one. If first is set, fail
+// if the pipe already exists.
+func createPipe(address string, first bool) (syscall.Handle, error) {
+ n, err := syscall.UTF16PtrFromString(address)
+ if err != nil {
+ return 0, err
+ }
+ mode := uint32(pipe_access_duplex | syscall.FILE_FLAG_OVERLAPPED)
+ if first {
+ mode |= file_flag_first_pipe_instance
+ }
+ return createNamedPipe(n,
+ mode,
+ pipe_type_byte,
+ pipe_unlimited_instances,
+ 512, 512, 0, nil)
+}
+
+func badAddr(addr string) PipeError {
+ return PipeError{fmt.Sprintf("Invalid pipe address '%s'.", addr), false}
+}
+func timeout(addr string) PipeError {
+ return PipeError{fmt.Sprintf("Pipe IO timed out waiting for '%s'", addr), true}
+}
+
+
+
+func newIpcClient(cfg IpcConfig, codec codec.Codec) (*ipcClient, error) {
+ c, err := Dial(cfg.Endpoint)
+ if err != nil {
+ return nil, err
+ }
+
+ return &ipcClient{codec.New(c)}, nil
+}
+
+func startIpc(cfg IpcConfig, codec codec.Codec, api api.Ethereum) error {
+ os.Remove(cfg.Endpoint) // in case it still exists from a previous run
+
+ l, err := Listen(cfg.Endpoint)
+ if err != nil {
+ return err
+ }
+ os.Chmod(cfg.Endpoint, 0600)
+
+ go func() {
+ for {
+ conn, err := l.Accept()
+ if err != nil {
+ glog.V(logger.Error).Infof("Error accepting ipc connection - %v\n", err)
+ continue
+ }
+
+ go func(conn net.Conn) {
+ codec := codec.New(conn)
+
+ for {
+ req, err := codec.ReadRequest()
+ if err == io.EOF {
+ codec.Close()
+ return
+ } else if err != nil {
+ glog.V(logger.Error).Infof("IPC recv err - %v\n", err)
+ codec.Close()
+ return
+ }
+
+ var rpcResponse interface{}
+ res, err := api.Execute(req)
+
+ rpcResponse = shared.NewRpcResponse(req.Id, req.Jsonrpc, res, err)
+ err = codec.WriteResponse(rpcResponse)
+ if err != nil {
+ glog.V(logger.Error).Infof("IPC send err - %v\n", err)
+ codec.Close()
+ return
+ }
+ }
+ }(conn)
+ }
+ }()
+
+ glog.V(logger.Info).Infof("IPC service started (%s)\n", cfg.Endpoint)
+
+ return nil
+}
diff --git a/rpc/shared/types.go b/rpc/shared/types.go
index 46fd5552c..600d39541 100644
--- a/rpc/shared/types.go
+++ b/rpc/shared/types.go
@@ -1,6 +1,10 @@
package shared
-import "encoding/json"
+import (
+ "encoding/json"
+ "github.com/ethereum/go-ethereum/logger"
+ "github.com/ethereum/go-ethereum/logger/glog"
+)
// RPC request
type Request struct {
@@ -36,3 +40,24 @@ type ErrorObject struct {
Message string `json:"message"`
// Data interface{} `json:"data"`
}
+
+func NewRpcResponse(id interface{}, jsonrpcver string, reply interface{}, err error) *interface{} {
+ var response interface{}
+
+ switch err.(type) {
+ case nil:
+ response = &SuccessResponse{Jsonrpc: jsonrpcver, Id: id, Result: reply}
+ case *NotImplementedError:
+ jsonerr := &ErrorObject{-32601, err.Error()}
+ response = &ErrorResponse{Jsonrpc: jsonrpcver, Id: id, Error: jsonerr}
+ case *DecodeParamError, *InsufficientParamsError, *ValidationError, *InvalidTypeError:
+ jsonerr := &ErrorObject{-32602, err.Error()}
+ response = &ErrorResponse{Jsonrpc: jsonrpcver, Id: id, Error: jsonerr}
+ default:
+ jsonerr := &ErrorObject{-32603, err.Error()}
+ response = &ErrorResponse{Jsonrpc: jsonrpcver, Id: id, Error: jsonerr}
+ }
+
+ glog.V(logger.Detail).Infof("Generated response: %T %s", response, response)
+ return &response
+}
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-16 20:50:07 +0800