-
Notifications
You must be signed in to change notification settings - Fork 661
/
server.go
141 lines (115 loc) · 3.74 KB
/
server.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
// (c) 2019-2020, Ava Labs, Inc. All rights reserved.
// See the file LICENSE for licensing terms.
package ipcs
import (
"fmt"
"net/http"
"nanomsg.org/go/mangos/v2/protocol/pub"
_ "nanomsg.org/go/mangos/v2/transport/ipc" // registers the IPC transport
"github.com/gorilla/rpc/v2"
"github.com/ava-labs/gecko/api"
"github.com/ava-labs/gecko/chains"
"github.com/ava-labs/gecko/snow/engine/common"
"github.com/ava-labs/gecko/snow/triggers"
"github.com/ava-labs/gecko/utils/json"
"github.com/ava-labs/gecko/utils/logging"
"github.com/ava-labs/gecko/utils/wrappers"
)
const baseURL = "ipc:///tmp/"
// IPCs maintains the IPCs
type IPCs struct {
log logging.Logger
chainManager chains.Manager
httpServer *api.Server
events *triggers.EventDispatcher
chains map[[32]byte]*ChainIPC
}
// NewService returns a new IPCs API service
func NewService(log logging.Logger, chainManager chains.Manager, events *triggers.EventDispatcher, httpServer *api.Server) *common.HTTPHandler {
newServer := rpc.NewServer()
codec := json.NewCodec()
newServer.RegisterCodec(codec, "application/json")
newServer.RegisterCodec(codec, "application/json;charset=UTF-8")
newServer.RegisterService(&IPCs{
log: log,
chainManager: chainManager,
httpServer: httpServer,
events: events,
chains: map[[32]byte]*ChainIPC{},
}, "ipcs")
return &common.HTTPHandler{Handler: newServer}
}
// PublishBlockchainArgs are the arguments for calling PublishBlockchain
type PublishBlockchainArgs struct {
BlockchainID string `json:"blockchainID"`
}
// PublishBlockchainReply are the results from calling PublishBlockchain
type PublishBlockchainReply struct {
URL string `json:"url"`
}
// PublishBlockchain publishes the finalized accepted transactions from the blockchainID over the IPC
func (ipc *IPCs) PublishBlockchain(r *http.Request, args *PublishBlockchainArgs, reply *PublishBlockchainReply) error {
chainID, err := ipc.chainManager.Lookup(args.BlockchainID)
if err != nil {
ipc.log.Error("unknown blockchainID: %s", err)
return err
}
chainIDKey := chainID.Key()
chainIDStr := chainID.String()
url := baseURL + chainIDStr + ".ipc"
reply.URL = url
if _, ok := ipc.chains[chainIDKey]; ok {
ipc.log.Info("returning existing blockchainID %s", chainIDStr)
return nil
}
sock, err := pub.NewSocket()
if err != nil {
ipc.log.Error("can't get new pub socket: %s", err)
return err
}
if err = sock.Listen(url); err != nil {
ipc.log.Error("can't listen on pub socket: %s", err)
sock.Close()
return err
}
chainIPC := &ChainIPC{
log: ipc.log,
socket: sock,
}
if err := ipc.events.RegisterChain(chainID, "ipc", chainIPC); err != nil {
ipc.log.Error("couldn't register event: %s", err)
sock.Close()
return err
}
ipc.chains[chainIDKey] = chainIPC
return nil
}
// UnpublishBlockchainArgs are the arguments for calling UnpublishBlockchain
type UnpublishBlockchainArgs struct {
BlockchainID string `json:"blockchainID"`
}
// UnpublishBlockchainReply are the results from calling UnpublishBlockchain
type UnpublishBlockchainReply struct {
Success bool `json:"success"`
}
// UnpublishBlockchain closes publishing of a blockchainID
func (ipc *IPCs) UnpublishBlockchain(r *http.Request, args *UnpublishBlockchainArgs, reply *UnpublishBlockchainReply) error {
chainID, err := ipc.chainManager.Lookup(args.BlockchainID)
if err != nil {
ipc.log.Error("unknown blockchainID %s: %s", args.BlockchainID, err)
return err
}
chainIDKey := chainID.Key()
chain, ok := ipc.chains[chainIDKey]
if !ok {
return fmt.Errorf("blockchainID not publishing: %s", chainID)
}
errs := wrappers.Errs{}
errs.Add(
chain.Stop(),
ipc.events.DeregisterChain(chainID, "ipc"),
)
delete(ipc.chains, chainIDKey)
reply.Success = true
return errs.Err
}