status-im
/
status-go
mirror of https://github.com/status-im/status-go.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

Use a `callReturn` type in `RoutineQueue`

This commit is contained in:
Pedro Pombeiro 2019-12-18 14:18:31 +01:00 committed by Pedro Pombeiro
parent 8d8880dc62
commit 90349b28c7
4 changed files with 106 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
eth-node/bridge/nimbus/node.go
View File

@ -75,14 +75,10 @@ func NewNodeBridge() Node {
}
func (n *nimbusNodeWrapper) StartNimbus(privateKey *ecdsa.PrivateKey, listenAddr string, staging bool) error {
retVal := n.routineQueue.Send(func(c chan<- interface{}) {
c <- startNimbus(privateKey, listenAddr, staging)
return n.routineQueue.Send(func(c chan<- callReturn) {
c <- callReturn{err: startNimbus(privateKey, listenAddr, staging)}
n.nodeStarted = true
})
if err, ok := retVal.(error); ok {
return err
}
return nil
}).err
}
func (n *nimbusNodeWrapper) Stop() {

52
eth-node/bridge/nimbus/public_whisper_api.go
View File

@ -40,63 +40,63 @@ func NewNimbusPublicWhisperAPIWrapper(filterMessagesMu *sync.Mutex, filterMessag
// AddPrivateKey imports the given private key.
func (w *nimbusPublicWhisperAPIWrapper) AddPrivateKey(ctx context.Context, privateKey types.HexBytes) (string, error) {
retVal := w.routineQueue.Send(func(c chan<- interface{}) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
privKeyC := C.CBytes(privateKey)
defer C.free(unsafe.Pointer(privKeyC))
idC := C.malloc(C.size_t(C.ID_LEN))
defer C.free(idC)
if C.nimbus_add_keypair((*C.uchar)(privKeyC), (*C.uchar)(idC)) {
c <- types.EncodeHex(C.GoBytes(idC, C.ID_LEN))
c <- callReturn{value: types.EncodeHex(C.GoBytes(idC, C.ID_LEN))}
} else {
c <- errors.New("failed to add private key to Nimbus")
c <- callReturn{err: errors.New("failed to add private key to Nimbus")}
}
})
if err, ok := retVal.(error); ok {
return "", err
if retVal.err != nil {
return "", retVal.err
}
return retVal.(string), nil
return retVal.value.(string), nil
}
// GenerateSymKeyFromPassword derives a key from the given password, stores it, and returns its ID.
func (w *nimbusPublicWhisperAPIWrapper) GenerateSymKeyFromPassword(ctx context.Context, passwd string) (string, error) {
retVal := w.routineQueue.Send(func(c chan<- interface{}) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
passwordC := C.CString(passwd)
defer C.free(unsafe.Pointer(passwordC))
idC := C.malloc(C.size_t(C.ID_LEN))
defer C.free(idC)
if C.nimbus_add_symkey_from_password(passwordC, (*C.uchar)(idC)) {
c <- types.EncodeHex(C.GoBytes(idC, C.ID_LEN))
c <- callReturn{value: types.EncodeHex(C.GoBytes(idC, C.ID_LEN))}
} else {
c <- errors.New("failed to add symkey to Nimbus")
c <- callReturn{err: errors.New("failed to add symkey to Nimbus")}
}
})
if err, ok := retVal.(error); ok {
return "", err
if retVal.err != nil {
return "", retVal.err
}
return retVal.(string), nil
return retVal.value.(string), nil
}
// DeleteKeyPair removes the key with the given key if it exists.
func (w *nimbusPublicWhisperAPIWrapper) DeleteKeyPair(ctx context.Context, key string) (bool, error) {
retVal := w.routineQueue.Send(func(c chan<- interface{}) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
keyC, err := decodeHexID(key)
if err != nil {
c <- err
c <- callReturn{err: err}
return
}
defer C.free(unsafe.Pointer(keyC))
c <- C.nimbus_delete_keypair(keyC)
c <- callReturn{value: C.nimbus_delete_keypair(keyC)}
})
if err, ok := retVal.(error); ok {
return false, err
if retVal.err != nil {
return false, retVal.err
}
return retVal.(bool), nil
return retVal.value.(bool), nil
}
// NewMessageFilter creates a new filter that can be used to poll for
@ -156,7 +156,7 @@ func (w *nimbusPublicWhisperAPIWrapper) GetFilterMessages(id string) ([]*types.M
// Post posts a message on the Whisper network.
// returns the hash of the message in case of success.
func (w *nimbusPublicWhisperAPIWrapper) Post(ctx context.Context, req types.NewMessage) ([]byte, error) {
retVal := w.routineQueue.Send(func(c chan<- interface{}) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
msg := C.post_message{
ttl: C.uint32_t(req.TTL),
powTime: C.double(req.PowTime),
@ -165,7 +165,7 @@ func (w *nimbusPublicWhisperAPIWrapper) Post(ctx context.Context, req types.NewM
if req.SigID != "" {
sourceID, err := decodeHexID(req.SigID)
if err != nil {
c <- err
c <- callReturn{err: err}
return
}
msg.sourceID = sourceID
@ -174,7 +174,7 @@ func (w *nimbusPublicWhisperAPIWrapper) Post(ctx context.Context, req types.NewM
if req.SymKeyID != "" {
symKeyID, err := decodeHexID(req.SymKeyID)
if err != nil {
c <- err
c <- callReturn{err: err}
return
}
msg.symKeyID = symKeyID
@ -194,19 +194,19 @@ func (w *nimbusPublicWhisperAPIWrapper) Post(ctx context.Context, req types.NewM
// TODO: return envelope hash once nimbus_post is improved to return it
if C.nimbus_post(&msg) {
c <- make([]byte, 0)
c <- callReturn{value: make([]byte, 0)}
return
}
c <- fmt.Errorf("failed to post message symkeyid=%s pubkey=%#x topic=%#x", req.SymKeyID, req.PublicKey, req.Topic[:])
c <- callReturn{err: fmt.Errorf("failed to post message symkeyid=%s pubkey=%#x topic=%#x", req.SymKeyID, req.PublicKey, req.Topic[:])}
// hashC := C.nimbus_post(&msg)
// if hashC == nil {
// return nil, errors.New("Nimbus failed to post message")
// }
// return hex.DecodeString(C.GoString(hashC))
})
if err, ok := retVal.(error); ok {
return nil, err
if retVal.err != nil {
return nil, retVal.err
}
return retVal.([]byte), nil
return retVal.value.([]byte), nil
}

13
eth-node/bridge/nimbus/routine_queue.go
View File

@ -14,6 +14,11 @@ type RoutineQueue struct {
events chan event
}
type callReturn struct {
value interface{}
err error
}
// NewRoutineQueue returns a new RoutineQueue object.
func NewRoutineQueue() *RoutineQueue {
q := &RoutineQueue{
@ -26,8 +31,8 @@ func NewRoutineQueue() *RoutineQueue {
// event represents an event triggered by the user.
type event struct {
f func(chan<- interface{})
done chan interface{}
f func(chan<- callReturn)
done chan callReturn
}
func (q *RoutineQueue) HandleEvent() {
@ -47,8 +52,8 @@ func (q *RoutineQueue) HandleEvent() {
// goroutine in order to execute a Nimbus function. It is important to note that the
// passed function won't be executed immediately, instead it will be added to
// the user events queue.
func (q *RoutineQueue) Send(f func(chan<- interface{})) interface{} {
ev := event{f: f, done: make(chan interface{}, 1)}
func (q *RoutineQueue) Send(f func(chan<- callReturn)) callReturn {
ev := event{f: f, done: make(chan callReturn, 1)}
defer close(ev.done)
if syscall.Gettid() == q.tid {
f(ev.done)

129
eth-node/bridge/nimbus/whisper.go
View File

@ -52,9 +52,9 @@ func (w *nimbusWhisperWrapper) PublicWhisperAPI() types.PublicWhisperAPI {
// MinPow returns the PoW value required by this node.
func (w *nimbusWhisperWrapper) MinPow() float64 {
return w.routineQueue.Send(func(c chan<- interface{}) {
c <- float64(C.nimbus_get_min_pow())
}).(float64)
return w.routineQueue.Send(func(c chan<- callReturn) {
c <- callReturn{value: float64(C.nimbus_get_min_pow())}
}).value.(float64)
}
// BloomFilter returns the aggregated bloom filter for all the topics of interest.
@ -62,7 +62,7 @@ func (w *nimbusWhisperWrapper) MinPow() float64 {
// If a message does not match the bloom, it will tantamount to spam, and the peer will
// be disconnected.
func (w *nimbusWhisperWrapper) BloomFilter() []byte {
return w.routineQueue.Send(func(c chan<- interface{}) {
return w.routineQueue.Send(func(c chan<- callReturn) {
// Allocate a buffer for Nimbus to return the bloom filter on
dataC := C.malloc(C.size_t(C.BLOOM_LEN))
defer C.free(unsafe.Pointer(dataC))
@ -72,8 +72,8 @@ func (w *nimbusWhisperWrapper) BloomFilter() []byte {
// Move the returned data into a Go array
data := make([]byte, C.BLOOM_LEN)
copy(data, C.GoBytes(dataC, C.BLOOM_LEN))
c <- data
}).([]byte)
c <- callReturn{value: data}
}).value.([]byte)
}
// GetCurrentTime returns current time.
@ -98,10 +98,10 @@ func (w *nimbusWhisperWrapper) SelectedKeyPairID() string {
}
func (w *nimbusWhisperWrapper) GetPrivateKey(id string) (*ecdsa.PrivateKey, error) {
retVal := w.routineQueue.Send(func(c chan<- interface{}) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
idC, err := decodeHexID(id)
if err != nil {
c <- err
c <- callReturn{err: err}
return
}
defer C.free(unsafe.Pointer(idC))
@ -109,28 +109,28 @@ func (w *nimbusWhisperWrapper) GetPrivateKey(id string) (*ecdsa.PrivateKey, erro
defer C.free(unsafe.Pointer(privKeyC))
if !C.nimbus_get_private_key(idC, (*C.uchar)(privKeyC)) {
c <- errors.New("failed to get private key from Nimbus")
c <- callReturn{err: errors.New("failed to get private key from Nimbus")}
return
}
pk, err := crypto.ToECDSA(C.GoBytes(privKeyC, C.PRIVKEY_LEN))
if err != nil {
c <- err
c <- callReturn{err: err}
return
}
c <- pk
c <- callReturn{value: pk}
})
if err, ok := retVal.(error); ok {
return nil, err
if retVal.err != nil {
return nil, retVal.err
}
return retVal.(*ecdsa.PrivateKey), nil
return retVal.value.(*ecdsa.PrivateKey), nil
}
// AddKeyPair imports a asymmetric private key and returns a deterministic identifier.
func (w *nimbusWhisperWrapper) AddKeyPair(key *ecdsa.PrivateKey) (string, error) {
retVal := w.routineQueue.Send(func(c chan<- interface{}) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
privKey := crypto.FromECDSA(key)
privKeyC := C.CBytes(privKey)
defer C.free(unsafe.Pointer(privKeyC))
@ -138,56 +138,61 @@ func (w *nimbusWhisperWrapper) AddKeyPair(key *ecdsa.PrivateKey) (string, error)
idC := C.malloc(C.size_t(C.ID_LEN))
defer C.free(idC)
if !C.nimbus_add_keypair((*C.uchar)(privKeyC), (*C.uchar)(idC)) {
c <- errors.New("failed to add keypair to Nimbus")
c <- callReturn{err: errors.New("failed to add keypair to Nimbus")}
return
}
c <- types.EncodeHex(C.GoBytes(idC, C.ID_LEN))
c <- callReturn{value: types.EncodeHex(C.GoBytes(idC, C.ID_LEN))}
})
if err, ok := retVal.(error); ok {
return "", err
if retVal.err != nil {
return "", retVal.err
}
return retVal.(string), nil
return retVal.value.(string), nil
}
// DeleteKeyPair deletes the key with the specified ID if it exists.
func (w *nimbusWhisperWrapper) DeleteKeyPair(keyID string) bool {
return w.routineQueue.Send(func(c chan<- interface{}) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
keyC, err := decodeHexID(keyID)
if err != nil {
c <- err
c <- callReturn{err: err}
return
}
defer C.free(unsafe.Pointer(keyC))
c <- C.nimbus_delete_keypair(keyC)
}).(bool)
c <- callReturn{value: C.nimbus_delete_keypair(keyC)}
})
if retVal.err != nil {
return false
}
return retVal.value.(bool)
}
func (w *nimbusWhisperWrapper) AddSymKeyDirect(key []byte) (string, error) {
retVal := w.routineQueue.Send(func(c chan<- interface{}) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
keyC := C.CBytes(key)
defer C.free(unsafe.Pointer(keyC))
idC := C.malloc(C.size_t(C.ID_LEN))
defer C.free(idC)
if !C.nimbus_add_symkey((*C.uchar)(keyC), (*C.uchar)(idC)) {
c <- errors.New("failed to add symkey to Nimbus")
c <- callReturn{err: errors.New("failed to add symkey to Nimbus")}
return
}
c <- types.EncodeHex(C.GoBytes(idC, C.ID_LEN))
c <- callReturn{value: types.EncodeHex(C.GoBytes(idC, C.ID_LEN))}
})
if err, ok := retVal.(error); ok {
return "", err
if retVal.err != nil {
return "", retVal.err
}
return retVal.(string), nil
return retVal.value.(string), nil
}
func (w *nimbusWhisperWrapper) AddSymKeyFromPassword(password string) (string, error) {
retVal := w.routineQueue.Send(func(c chan<- interface{}) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
passwordC := C.CString(password)
defer C.free(unsafe.Pointer(passwordC))
@ -195,36 +200,41 @@ func (w *nimbusWhisperWrapper) AddSymKeyFromPassword(password string) (string, e
defer C.free(idC)
if C.nimbus_add_symkey_from_password(passwordC, (*C.uchar)(idC)) {
id := C.GoBytes(idC, C.ID_LEN)
c <- types.EncodeHex(id)
c <- callReturn{value: types.EncodeHex(id)}
} else {
c <- errors.New("failed to add symkey to Nimbus")
c <- callReturn{err: errors.New("failed to add symkey to Nimbus")}
}
})
if err, ok := retVal.(error); ok {
return "", err
if retVal.err != nil {
return "", retVal.err
}
return retVal.(string), nil
return retVal.value.(string), nil
}
func (w *nimbusWhisperWrapper) DeleteSymKey(id string) bool {
return w.routineQueue.Send(func(c chan<- interface{}) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
idC, err := decodeHexID(id)
if err != nil {
c <- err
c <- callReturn{err: err}
return
}
defer C.free(unsafe.Pointer(idC))
c <- C.nimbus_delete_symkey(idC)
}).(bool)
c <- callReturn{value: C.nimbus_delete_symkey(idC)}
})
if retVal.err != nil {
return false
}
return retVal.value.(bool)
}
func (w *nimbusWhisperWrapper) GetSymKey(id string) ([]byte, error) {
retVal := w.routineQueue.Send(func(c chan<- interface{}) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
idC, err := decodeHexID(id)
if err != nil {
c <- err
c <- callReturn{err: err}
return
}
defer C.free(unsafe.Pointer(idC))
@ -233,17 +243,17 @@ func (w *nimbusWhisperWrapper) GetSymKey(id string) ([]byte, error) {
dataC := C.malloc(C.size_t(C.SYMKEY_LEN))
defer C.free(unsafe.Pointer(dataC))
if !C.nimbus_get_symkey(idC, (*C.uchar)(dataC)) {
c <- errors.New("symkey not found")
c <- callReturn{err: errors.New("symkey not found")}
return
}
c <- C.GoBytes(dataC, C.SYMKEY_LEN)
c <- callReturn{value: C.GoBytes(dataC, C.SYMKEY_LEN)}
})
if err, ok := retVal.(error); ok {
return nil, err
if retVal.err != nil {
return nil, retVal.err
}
return retVal.([]byte), nil
return retVal.value.([]byte), nil
}
//export onMessageHandler
@ -277,7 +287,7 @@ func (w *nimbusWhisperWrapper) Subscribe(opts *types.SubscriptionOptions) (strin
return "", err
}
retVal := w.routineQueue.Send(func(c chan<- interface{}) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
// Create a message store for this filter, so we can add new messages to it from the nimbus_subscribe_filter callback
messageList := list.New()
idC := C.malloc(C.size_t(C.ID_LEN))
@ -286,7 +296,7 @@ func (w *nimbusWhisperWrapper) Subscribe(opts *types.SubscriptionOptions) (strin
GetNimbusFilterFrom(f),
(C.received_msg_handler)(unsafe.Pointer(C.onMessageHandler_cgo)), gopointer.Save(messageList),
(*C.uchar)(idC)) {
c <- errors.New("failed to subscribe to filter in Nimbus")
c <- callReturn{err: errors.New("failed to subscribe to filter in Nimbus")}
return
}
filterID := C.GoString((*C.char)(idC))
@ -297,13 +307,13 @@ func (w *nimbusWhisperWrapper) Subscribe(opts *types.SubscriptionOptions) (strin
f.(*nimbusFilterWrapper).id = filterID
c <- filterID
c <- callReturn{value: filterID}
})
if err, ok := retVal.(error); ok {
return "", err
if retVal.err != nil {
return "", retVal.err
}
return retVal.(string), nil
return retVal.value.(string), nil
}
func (w *nimbusWhisperWrapper) GetFilter(id string) types.Filter {
@ -316,16 +326,16 @@ func (w *nimbusWhisperWrapper) GetFilter(id string) types.Filter {
}
func (w *nimbusWhisperWrapper) Unsubscribe(id string) error {
retVal := w.routineQueue.Send(func(c chan<- interface{}) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
idC, err := decodeHexID(id)
if err != nil {
c <- err
c <- callReturn{err: err}
return
}
defer C.free(unsafe.Pointer(idC))
if ok := C.nimbus_unsubscribe_filter(idC); !ok {
c <- errors.New("filter not found")
c <- callReturn{err: errors.New("filter not found")}
return
}
@ -341,12 +351,9 @@ func (w *nimbusWhisperWrapper) Unsubscribe(id string) error {
delete(w.filters, id)
}
c <- nil
c <- callReturn{err: nil}
})
if err, ok := retVal.(error); ok {
return err
}
return nil
return retVal.err
}
func decodeHexID(id string) (*C.uint8_t, error) {