Implement most of eth_subscribe in a websocket

pkg/eth/eth_address.go

@@ -0,0 +1,60 @@
+package eth
+
+import (
+	"encoding/json"
+	"strings"
+
+	"github.com/pkg/errors"
+
+	"github.com/ethereum/go-ethereum/common/hexutil"
+)
+
+var ErrNoHexPrefix = errors.New("Missing 0x prefix")
+var ErrInvalidLength = errors.New("Invalid length")
+
+type ETHAddress struct {
+	address string
+}
+
+func (addr *ETHAddress) String() string {
+	return addr.address
+}
+
+func (addr ETHAddress) MarshalJSON() ([]byte, error) {
+	if err := validateAddress(addr.address); err != nil {
+		return []byte{}, err
+	}
+
+	return json.Marshal(addr.address)
+}
+
+// UnmarshalJSON needs to be able to parse ETHAddress from both hex string or number
+func (addr *ETHAddress) UnmarshalJSON(data []byte) (err error) {
+	asString := string(data)
+	if strings.HasPrefix(asString, `"`) && strings.HasSuffix(asString, `"`) {
+		asString = asString[1 : len(asString)-1]
+	}
+	if err := validateAddress(asString); err != nil {
+		return err
+	}
+
+	addr.address = asString
+	return nil
+}
+
+func validateAddress(address string) error {
+	if !strings.HasPrefix(address, "0x") {
+		return ErrNoHexPrefix
+	}
+
+	if len(address) != 42 {
+		return ErrInvalidLength
+	}
+
+	_, err := hexutil.Decode(address)
+	if err != nil {
+		return errors.Wrap(err, "Invalid hexadecimal")
+	}
+
+	return nil
+}

pkg/eth/rpc_types.go

@@ -635,6 +635,60 @@ func newErrInvalidParameterType(idx int, gotType interface{}, wantedType interfa
 	return errors.Errorf("invalid %d parameter of %T type, but %T type is expected", idx, gotType, wantedType)
 }
 
+// ========== eth_subscribe ============= //
+
+type (
+	EthLogSubscriptionParameter struct {
+		Address ETHAddress    `json:"address"`
+		Topics  []interface{} `json:"topics"`
+	}
+
+	EthSubscriptionRequest struct {
+		Method string
+		Params *EthLogSubscriptionParameter
+	}
+
+	EthSubscriptionResponse string
+)
+
+func (r *EthSubscriptionRequest) UnmarshalJSON(data []byte) error {
+	var params []interface{}
+	if err := json.Unmarshal(data, &params); err != nil {
+		return errors.Wrap(err, "couldn't unmarhsal data")
+	}
+
+	method, ok := params[0].(string)
+	if !ok {
+		return newErrInvalidParameterType(1, params[0], "")
+	}
+	r.Method = method
+
+	if len(params) >= 2 {
+		param, err := json.Marshal(params[1])
+		if err != nil {
+			return err
+		}
+		var subscriptionParameter EthLogSubscriptionParameter
+		err = json.Unmarshal(param, &subscriptionParameter)
+		if err != nil {
+			return err
+		}
+		r.Params = &subscriptionParameter
+	}
+
+	return nil
+}
+
+func (r EthSubscriptionRequest) MarshalJSON() ([]byte, error) {
+	output := []interface{}{}
+	output = append(output, r.Method)
+	if r.Params != nil {
+		output = append(output, r.Params)
+	}
+
+	return json.Marshal(output)
+}
+
 // ========== eth_newFilter ============= //
 
 type NewFilterRequest struct {
@@ -708,6 +762,12 @@ func (r *GetStorageRequest) UnmarshalJSON(data []byte) error {
 // ======= eth_chainId ============= //
 type ChainIdResponse string
 
+// ======= eth_subscription ======== //
+type EthSubscription struct {
+	SubscriptionID string      `json:"subscription"`
+	Result         interface{} `json:"result"`
+}
+
 // ======= qtum_getUTXOs ============= //
 
 type (

pkg/eth/rpc_types_test.go

@@ -0,0 +1,47 @@
+package eth
+
+import (
+	"encoding/json"
+	"testing"
+)
+
+func TestEthLogSubscriptionRequestSerialization(t *testing.T) {
+	jsonValue := `["logs",{"address":"0x8320fe7702b96808f7bbc0d4a888ed1468216cfd","topics":["0xd78a0cb8bb633d06981248b816e7bd33c2a35a6089241d099fa519e361cab902"]}]`
+	var request EthSubscriptionRequest
+	err := json.Unmarshal([]byte(jsonValue), &request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	asJson, err := json.Marshal(request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(asJson) != jsonValue {
+		t.Fatalf(`"%s" != "%s"\n`, string(asJson), jsonValue)
+	}
+}
+
+func TestEthLogSubscriptionRequestWithInvalidAddressSerialization(t *testing.T) {
+	jsonValue := `["logs",{"address":"0x0","topics":["0xd78a0cb8bb633d06981248b816e7bd33c2a35a6089241d099fa519e361cab902"]}]`
+	var request EthSubscriptionRequest
+	err := json.Unmarshal([]byte(jsonValue), &request)
+	if err != ErrInvalidLength {
+		t.Fatal(err)
+	}
+}
+
+func TestEthNewPendingTransactionsRequestSerialization(t *testing.T) {
+	jsonValue := `["newPendingTransactions"]`
+	var request EthSubscriptionRequest
+	err := json.Unmarshal([]byte(jsonValue), &request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	asJson, err := json.Marshal(request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(asJson) != jsonValue {
+		t.Fatalf(`"%s" != "%s"\n`, string(asJson), jsonValue)
+	}
+}

pkg/eth/util.go

@@ -0,0 +1,35 @@
+package eth
+
+import (
+	"github.com/pkg/errors"
+	"github.com/qtumproject/janus/pkg/utils"
+)
+
+// translateTopics takes in an ethReq's topics field and translates it to a it's equivalent QtumReq
+// topics (optional) has a max lenght of 4
+func TranslateTopics(ethTopics []interface{}) ([]interface{}, error) {
+
+	var topics []interface{}
+
+	if len(ethTopics) > 4 {
+		return nil, errors.Errorf("invalid number of topics. Logs have a max of 4 topics.")
+	}
+
+	for _, topic := range ethTopics {
+		switch topic.(type) {
+		case []interface{}:
+			topic, err := TranslateTopics(topic.([]interface{}))
+			if err != nil {
+				return nil, err
+			}
+			topics = append(topics, topic)
+		case string:
+			topics = append(topics, utils.RemoveHexPrefix(topic.(string)))
+		case nil:
+			topics = append(topics, nil)
+		}
+	}
+
+	return topics, nil
+
+}