clients/horizonclient: support dynamic effects

clients/horizonclient/CHANGELOG.md

@@ -10,6 +10,8 @@ file.  This project adheres to [Semantic Versioning](http://semver.org/).
 
 - `Client.Root()` method for querying the root endpoint of a horizon server.
 
-### Changes 
+### Changes
 
 - `Client.Fund()` now returns `TransactionSuccess` instead of a http response pointer.
+
+- Querying the effects endpoint now supports returning the concrete effect type for each effect. This is also supported in streaming mode. See the [docs](https://godoc.org/github.com/stellar/go/clients/horizonclient#Client.Effects) for examples.

clients/horizonclient/client.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/manucorporat/sse"
 	hProtocol "github.com/stellar/go/protocols/horizon"
+	"github.com/stellar/go/protocols/horizon/effects"
 	"github.com/stellar/go/protocols/horizon/operations"
 	"github.com/stellar/go/support/app"
 	"github.com/stellar/go/support/errors"
@@ -251,7 +252,7 @@ func (c *Client) AccountData(request AccountRequest) (accountData hProtocol.Acco
 
 // Effects returns effects(https://www.stellar.org/developers/horizon/reference/resources/effect.html)
 // It can be used to return effects for an account, a ledger, an operation, a transaction and all effects on the network.
-func (c *Client) Effects(request EffectRequest) (effects hProtocol.EffectsPage, err error) {
+func (c *Client) Effects(request EffectRequest) (effects effects.EffectsPage, err error) {
 	err = c.sendRequest(request, &effects)
 	return
 }

clients/horizonclient/effect_request.go

@@ -11,59 +11,38 @@ import (
 )
 
 // EffectHandler is a function that is called when a new effect is received
-type EffectHandler func(effects.Base)
+type EffectHandler func(effects.Effect)
 
 // BuildURL creates the endpoint to be queried based on the data in the EffectRequest struct.
 // If no data is set, it defaults to the build the URL for all effects
 func (er EffectRequest) BuildURL() (endpoint string, err error) {
-
 	nParams := countParams(er.ForAccount, er.ForLedger, er.ForOperation, er.ForTransaction)
 
 	if nParams > 1 {
-		err = errors.New("invalid request: too many parameters")
-	}
-
-	if err != nil {
-		return endpoint, err
+		return endpoint, errors.New("invalid request: too many parameters")
 	}
 
 	endpoint = "effects"
 
 	if er.ForAccount != "" {
-		endpoint = fmt.Sprintf(
-			"accounts/%s/effects",
-			er.ForAccount,
-		)
+		endpoint = fmt.Sprintf("accounts/%s/effects", er.ForAccount)
 	}
 
 	if er.ForLedger != "" {
-		endpoint = fmt.Sprintf(
-			"ledgers/%s/effects",
-			er.ForLedger,
-		)
+		endpoint = fmt.Sprintf("ledgers/%s/effects", er.ForLedger)
 	}
 
 	if er.ForOperation != "" {
-		endpoint = fmt.Sprintf(
-			"operations/%s/effects",
-			er.ForOperation,
-		)
+		endpoint = fmt.Sprintf("operations/%s/effects", er.ForOperation)
 	}
 
 	if er.ForTransaction != "" {
-		endpoint = fmt.Sprintf(
-			"transactions/%s/effects",
-			er.ForTransaction,
-		)
+		endpoint = fmt.Sprintf("transactions/%s/effects", er.ForTransaction)
 	}
 
 	queryParams := addQueryParams(cursor(er.Cursor), limit(er.Limit), er.Order)
 	if queryParams != "" {
-		endpoint = fmt.Sprintf(
-			"%s?%s",
-			endpoint,
-			queryParams,
-		)
+		endpoint = fmt.Sprintf("%s?%s", endpoint, queryParams)
 	}
 
 	_, err = url.Parse(endpoint)
@@ -77,24 +56,27 @@ func (er EffectRequest) BuildURL() (endpoint string, err error) {
 // StreamEffects streams horizon effects. It can be used to stream all effects or account specific effects.
 // Use context.WithCancel to stop streaming or context.Background() if you want to stream indefinitely.
 // EffectHandler is a user-supplied function that is executed for each streamed effect received.
-func (er EffectRequest) StreamEffects(
-	ctx context.Context,
-	client *Client,
-	handler EffectHandler,
-) (err error) {
+func (er EffectRequest) StreamEffects(ctx context.Context, client *Client, handler EffectHandler) error {
 	endpoint, err := er.BuildURL()
 	if err != nil {
-		return errors.Wrap(err, "unable to build endpoint")
+		return errors.Wrap(err, "unable to build endpoint for effects request")
 	}
 
 	url := fmt.Sprintf("%s%s", client.fixHorizonURL(), endpoint)
 	return client.stream(ctx, url, func(data []byte) error {
-		var effect effects.Base
-		err = json.Unmarshal(data, &effect)
+		var baseEffect effects.Base
+		// unmarshal into the base effect type
+		if err = json.Unmarshal(data, &baseEffect); err != nil {
+			return errors.Wrap(err, "error unmarshaling data for effects request")
+		}
+
+		// unmarshal into the concrete effect type
+		effs, err := effects.UnmarshalEffect(baseEffect.GetType(), data)
 		if err != nil {
-			return errors.Wrap(err, "error unmarshaling data")
+			return errors.Wrap(err, "unmarshaling to the correct effect type")
 		}
-		handler(effect)
+
+		handler(effs)
 		return nil
 	})
 }

clients/horizonclient/effect_request_test.go

@@ -76,7 +76,7 @@ func ExampleClient_StreamEffects() {
 		cancel()
 	}()
 
-	printHandler := func(e effects.Base) {
+	printHandler := func(e effects.Effect) {
 		fmt.Println(e)
 	}
 	err := client.StreamEffects(ctx, effectRequest, printHandler)
@@ -100,14 +100,14 @@ func TestEffectRequestStreamEffects(t *testing.T) {
 		"https://localhost/effects?cursor=now",
 	).ReturnString(200, effectStreamResponse)
 
-	effectStream := make([]effects.Base, 1)
-	err := client.StreamEffects(ctx, effectRequest, func(effect effects.Base) {
+	effectStream := make([]effects.Effect, 1)
+	err := client.StreamEffects(ctx, effectRequest, func(effect effects.Effect) {
 		effectStream[0] = effect
 		cancel()
 	})
 
 	if assert.NoError(t, err) {
-		assert.Equal(t, effectStream[0].Type, "account_credited")
+		assert.Equal(t, effectStream[0].GetType(), "account_credited")
 	}
 
 	// Account effects
@@ -119,13 +119,13 @@ func TestEffectRequestStreamEffects(t *testing.T) {
 		"https://localhost/accounts/GBNZN27NAOHRJRCMHQF2ZN2F6TAPVEWKJIGZIRNKIADWIS2HDENIS6CI/effects?cursor=now",
 	).ReturnString(200, effectStreamResponse)
 
-	err = client.StreamEffects(ctx, effectRequest, func(effect effects.Base) {
+	err = client.StreamEffects(ctx, effectRequest, func(effect effects.Effect) {
 		effectStream[0] = effect
 		cancel()
 	})
 
 	if assert.NoError(t, err) {
-		assert.Equal(t, effectStream[0].Account, "GBNZN27NAOHRJRCMHQF2ZN2F6TAPVEWKJIGZIRNKIADWIS2HDENIS6CI")
+		assert.Equal(t, effectStream[0].GetAccount(), "GBNZN27NAOHRJRCMHQF2ZN2F6TAPVEWKJIGZIRNKIADWIS2HDENIS6CI")
 	}
 
 	// test error
@@ -137,7 +137,7 @@ func TestEffectRequestStreamEffects(t *testing.T) {
 		"https://localhost/effects?cursor=now",
 	).ReturnString(500, effectStreamResponse)
 
-	err = client.StreamEffects(ctx, effectRequest, func(effect effects.Base) {
+	err = client.StreamEffects(ctx, effectRequest, func(effect effects.Effect) {
 		effectStream[0] = effect
 		cancel()
 	})

clients/horizonclient/main.go

@@ -18,6 +18,7 @@ import (
 	"time"
 
 	hProtocol "github.com/stellar/go/protocols/horizon"
+	"github.com/stellar/go/protocols/horizon/effects"
 	"github.com/stellar/go/protocols/horizon/operations"
 	"github.com/stellar/go/support/render/problem"
 	"github.com/stellar/go/txnbuild"
@@ -122,7 +123,7 @@ type Client struct {
 type ClientInterface interface {
 	AccountDetail(request AccountRequest) (hProtocol.Account, error)
 	AccountData(request AccountRequest) (hProtocol.AccountData, error)
-	Effects(request EffectRequest) (hProtocol.EffectsPage, error)
+	Effects(request EffectRequest) (effects.EffectsPage, error)
 	Assets(request AssetRequest) (hProtocol.AssetsPage, error)
 	Ledgers(request LedgerRequest) (hProtocol.LedgersPage, error)
 	LedgerDetail(sequence uint32) (hProtocol.Ledger, error)

clients/horizonclient/main_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/stellar/go/txnbuild"
 
 	hProtocol "github.com/stellar/go/protocols/horizon"
+	"github.com/stellar/go/protocols/horizon/effects"
 	"github.com/stellar/go/protocols/horizon/operations"
 	"github.com/stellar/go/support/errors"
 	"github.com/stellar/go/support/http/httptest"
@@ -47,7 +48,14 @@ func ExampleClient_Effects() {
 		fmt.Println(err)
 		return
 	}
-	fmt.Print(effect)
+	records := effect.Embedded.Records
+	if records[0].GetType() == "account_created" {
+		acc, ok := records[0].(effects.AccountCreated)
+		if ok {
+			fmt.Print(acc.Account)
+			fmt.Print(acc.StartingBalance)
+		}
+	}
 }
 
 func ExampleClient_Assets() {
@@ -496,10 +504,30 @@ func TestEffectsRequest(t *testing.T) {
 		"https://localhost/effects",
 	).ReturnString(200, effectsResponse)
 
-	effects, err := client.Effects(effectRequest)
+	effs, err := client.Effects(effectRequest)
 	if assert.NoError(t, err) {
-		assert.IsType(t, effects, hProtocol.EffectsPage{})
+		assert.IsType(t, effs, effects.EffectsPage{})
+		links := effs.Links
+		assert.Equal(t, links.Self.Href, "https://horizon-testnet.stellar.org/operations/43989725060534273/effects?cursor=&limit=10&order=asc")
+
+		assert.Equal(t, links.Next.Href, "https://horizon-testnet.stellar.org/operations/43989725060534273/effects?cursor=43989725060534273-3&limit=10&order=asc")
 
+		assert.Equal(t, links.Prev.Href, "https://horizon-testnet.stellar.org/operations/43989725060534273/effects?cursor=43989725060534273-1&limit=10&order=desc")
+
+		adEffect := effs.Embedded.Records[0]
+		acEffect := effs.Embedded.Records[1]
+		arEffect := effs.Embedded.Records[2]
+		assert.IsType(t, adEffect, effects.AccountDebited{})
+		assert.IsType(t, acEffect, effects.AccountCredited{})
+		// account_removed effect does not have a struct. Defaults to effects.Base
+		assert.IsType(t, arEffect, effects.Base{})
+
+		c, ok := acEffect.(effects.AccountCredited)
+		assert.Equal(t, ok, true)
+		assert.Equal(t, c.ID, "0043989725060534273-0000000002")
+		assert.Equal(t, c.Amount, "9999.9999900")
+		assert.Equal(t, c.Account, "GBO7LQUWCC7M237TU2PAXVPOLLYNHYCYYFCLVMX3RBJCML4WA742X3UB")
+		assert.Equal(t, c.Asset.Type, "native")
 	}
 
 	effectRequest = EffectRequest{ForAccount: "GCLWGQPMKXQSPF776IU33AH4PZNOOWNAWGGKVTBQMIC5IMKUNP3E6NVU"}
@@ -508,9 +536,9 @@ func TestEffectsRequest(t *testing.T) {
 		"https://localhost/accounts/GCLWGQPMKXQSPF776IU33AH4PZNOOWNAWGGKVTBQMIC5IMKUNP3E6NVU/effects",
 	).ReturnString(200, effectsResponse)
 
-	effects, err = client.Effects(effectRequest)
+	effs, err = client.Effects(effectRequest)
 	if assert.NoError(t, err) {
-		assert.IsType(t, effects, hProtocol.EffectsPage{})
+		assert.IsType(t, effs, effects.EffectsPage{})
 	}
 
 	// too many parameters
@@ -525,7 +553,6 @@ func TestEffectsRequest(t *testing.T) {
 	if assert.Error(t, err) {
 		assert.Contains(t, err.Error(), "too many parameters")
 	}
-
 }
 
 func TestAssetsRequest(t *testing.T) {

clients/horizonclient/mocks.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	hProtocol "github.com/stellar/go/protocols/horizon"
+	"github.com/stellar/go/protocols/horizon/effects"
 	"github.com/stellar/go/protocols/horizon/operations"
 	"github.com/stellar/go/txnbuild"
 	"github.com/stretchr/testify/mock"
@@ -27,9 +28,9 @@ func (m *MockClient) AccountData(request AccountRequest) (hProtocol.AccountData,
 }
 
 // Effects is a mocking method
-func (m *MockClient) Effects(request EffectRequest) (hProtocol.EffectsPage, error) {
+func (m *MockClient) Effects(request EffectRequest) (effects.EffectsPage, error) {
 	a := m.Called(request)
-	return a.Get(0).(hProtocol.EffectsPage), a.Error(1)
+	return a.Get(0).(effects.EffectsPage), a.Error(1)
 }
 
 // Assets is a mocking method