rpcclient: add getzmqnotifications RPC

btcjson/chainsvrcmds_test.go

@@ -1460,6 +1460,18 @@ func TestChainSvrCmds(t *testing.T) {
 			marshalled:   `{"jsonrpc":"1.0","method":"getdescriptorinfo","params":["123"],"id":1}`,
 			unmarshalled: &btcjson.GetDescriptorInfoCmd{Descriptor: "123"},
 		},
+		{
+			name: "getzmqnotifications",
+			newCmd: func() (interface{}, error) {
+				return btcjson.NewCmd("getzmqnotifications")
+			},
+			staticCmd: func() interface{} {
+				return btcjson.NewGetZmqNotificationsCmd()
+			},
+
+			marshalled:   `{"jsonrpc":"1.0","method":"getzmqnotifications","params":[],"id":1}`,
+			unmarshalled: &btcjson.GetZmqNotificationsCmd{},
+		},
 	}
 
 	t.Logf("Running %d tests", len(tests))

btcjson/chainsvrresults_test.go

@@ -6,12 +6,13 @@ package btcjson_test
 
 import (
 	"encoding/json"
+	"net/url"
 	"reflect"
 	"testing"
 
 	"github.com/btcsuite/btcd/btcjson"
-	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/btcutil"
+	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/davecgh/go-spew/spew"
 )
 
@@ -72,6 +73,21 @@ func TestChainSvrCustomResults(t *testing.T) {
 			},
 			expected: `{"txid":"123","vout":1,"scriptSig":{"asm":"0","hex":"00"},"prevOut":{"addresses":["addr1"],"value":0},"sequence":4294967295}`,
 		},
+		{
+			name: "zmq notification",
+			result: &btcjson.GetZmqNotificationResult{{
+				Type: "pubrawblock",
+				Address: func() *url.URL {
+					u, err := url.Parse("tcp://127.0.0.1:1238")
+					if err != nil {
+						panic(err)
+					}
+					return u
+				}(),
+				HighWaterMark: 1337,
+			}},
+			expected: `[{"address":"tcp://127.0.0.1:1238","hwm":1337,"type":"pubrawblock"}]`,
+		},
 	}
 
 	t.Logf("Running %d tests", len(tests))

btcjson/zmqsvrcmds.go

@@ -0,0 +1,16 @@
+package btcjson
+
+// GetZmqNotificationsCmd defines the getzmqnotifications JSON-RPC command.
+type GetZmqNotificationsCmd struct{}
+
+// NewGetZmqNotificationsCmd returns a new instance which can be used to issue a
+// getzmqnotifications JSON-RPC command.
+func NewGetZmqNotificationsCmd() *GetZmqNotificationsCmd {
+	return &GetZmqNotificationsCmd{}
+}
+
+func init() {
+	flags := UsageFlag(0)
+
+	MustRegisterCmd("getzmqnotifications", (*GetZmqNotificationsCmd)(nil), flags)
+}

btcjson/zmqsvrresults.go

@@ -0,0 +1,63 @@
+package btcjson
+
+import (
+	"encoding/json"
+	"net/url"
+)
+
+// GetZmqNotificationResult models the data returned from the getzmqnotifications command.
+type GetZmqNotificationResult []struct {
+	Type          string   // Type of notification
+	Address       *url.URL // Address of the publisher
+	HighWaterMark int      // Outbound message high water mark
+}
+
+func (z *GetZmqNotificationResult) MarshalJSON() ([]byte, error) {
+	var out []map[string]interface{}
+	for _, notif := range *z {
+		out = append(out,
+			map[string]interface{}{
+				"type":    notif.Type,
+				"address": notif.Address.String(),
+				"hwm":     notif.HighWaterMark,
+			})
+	}
+	return json.Marshal(out)
+}
+
+// UnmarshalJSON satisfies the json.Unmarshaller interface
+func (z *GetZmqNotificationResult) UnmarshalJSON(bytes []byte) error {
+	type basicNotification struct {
+		Type    string
+		Address string
+		Hwm     int
+	}
+
+	var basics []basicNotification
+	if err := json.Unmarshal(bytes, &basics); err != nil {
+		return err
+	}
+
+	var notifications GetZmqNotificationResult
+	for _, basic := range basics {
+
+		address, err := url.Parse(basic.Address)
+		if err != nil {
+			return err
+		}
+
+		notifications = append(notifications, struct {
+			Type          string
+			Address       *url.URL
+			HighWaterMark int
+		}{
+			Type:          basic.Type,
+			Address:       address,
+			HighWaterMark: basic.Hwm,
+		})
+	}
+
+	*z = notifications
+
+	return nil
+}

rpcclient/zmq.go

@@ -0,0 +1,39 @@
+package rpcclient
+
+import (
+	"encoding/json"
+
+	"github.com/btcsuite/btcd/btcjson"
+)
+
+// FutureGetZmqNotificationsResult is a future promise to deliver the result of
+// a GetZmqNotifications RPC invocation
+type FutureGetZmqNotificationsResult chan *Response
+
+// Receive waits for the response promised by the future and returns the unmarshalled
+// response, or an error if the request was unsuccessful.
+func (r FutureGetZmqNotificationsResult) Receive() (btcjson.GetZmqNotificationResult, error) {
+	res, err := ReceiveFuture(r)
+	if err != nil {
+		return nil, err
+	}
+	var notifications btcjson.GetZmqNotificationResult
+	if err := json.Unmarshal(res, &notifications); err != nil {
+		return nil, err
+	}
+	return notifications, nil
+}
+
+// GetZmqNotificationsAsync returns an instance ofa type that can be used to get
+// the result of a custom RPC request at some future time by invoking the Receive
+// function on the returned instance.
+//
+// See GetZmqNotifications for the blocking version and more details.
+func (c *Client) GetZmqNotificationsAsync() FutureGetZmqNotificationsResult {
+	return c.SendCmd(btcjson.NewGetZmqNotificationsCmd())
+}
+
+// GetZmqNotifications returns information about the active ZeroMQ notifications.
+func (c *Client) GetZmqNotifications() (btcjson.GetZmqNotificationResult, error) {
+	return c.GetZmqNotificationsAsync().Receive()
+}