Added Account flag decoding support

cmd/slnc/cmd/serum_get_market.go

@@ -64,6 +64,21 @@ var serumGetMarketCmd = &cobra.Command{
 		output = append(output, "------- | --------")
 		output = append(output, outputOrderBook(bids, totalSize, false)...)
 		output = append(output, "Bids")
+
+		fmt.Println(market.Name)
+
+		fmt.Println("Request RequestQueue: ", market.MarketV2.RequestQueue)
+		fmt.Println("Event RequestQueue: ", market.MarketV2.EventQueue)
+
+		fmt.Println("Base")
+		fmt.Println("base mint", market.MarketV2.BaseMint.String())
+		fmt.Println("base lot size", market.MarketV2.BaseLotSize)
+
+		fmt.Println("")
+		fmt.Println("Quote")
+		fmt.Println("quote mint", market.MarketV2.QuoteMint.String())
+		fmt.Println("quote lot size", market.MarketV2.QuoteLotSize)
+
 		fmt.Println(columnize.Format(output, nil))
 		return nil
 	},

programs/serum/instruction_test.go

@@ -2,9 +2,10 @@ package serum
 
 import (
 	"encoding/hex"
-	"fmt"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+
 	bin "github.com/dfuse-io/binary"
 	"github.com/stretchr/testify/require"
 )
@@ -16,9 +17,5 @@ func TestDecodeInstruction(t *testing.T) {
 	var instruction *Instruction
 	err = bin.NewDecoder(data).Decode(&instruction)
 	require.NoError(t, err)
-	fmt.Println(instruction)
-}
-
-func TestString(t *testing.T) {
-
+	assert.Equal(t, instruction.Version, uint8(0))
 }

programs/serum/queue.go

@@ -0,0 +1,109 @@
+// Copyright 2020 dfuse Platform Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package serum
+
+import (
+	bin "github.com/dfuse-io/binary"
+	"github.com/dfuse-io/solana-go"
+)
+
+type RequestQueue struct {
+	SerumPadding [5]byte `json:"-"`
+
+	AccountFlags AccountFlag
+	Head         bin.Uint64
+	Count        bin.Uint64 `bin:"sizeof=Requests"`
+	NextSeqNum   bin.Uint64
+	Requests     []*Request
+
+	EndPadding [7]byte `json:"-"`
+}
+
+func (r *RequestQueue) Decode(data []byte) error {
+	decoder := bin.NewDecoder(data)
+	return decoder.Decode(&r)
+}
+
+type Request struct {
+	RequestFlags         uint8
+	OwnerSlot            uint8
+	FeeTier              uint8
+	SelfTradeBehavior    uint8
+	Padding              [4]byte `json:"-"`
+	MaxCoinQtyOrCancelId bin.Uint64
+	NativePCQtyLocked    bin.Uint64
+	OrderId              bin.Uint128
+	Owner                [4]bin.Uint64
+	ClientOrderID        bin.Uint64
+}
+
+type EventQueue struct {
+	SerumPadding [5]byte `json:"-"`
+
+	AccountFlags AccountFlag
+	Head         bin.Uint64
+	Count        bin.Uint64 `bin:"sizeof=Events"`
+	SeqNum       bin.Uint64
+	Events       []*Event
+
+	EndPadding [7]byte `json:"-"`
+}
+
+func (q *EventQueue) Decode(data []byte) error {
+	decoder := bin.NewDecoder(data)
+	return decoder.Decode(&q)
+}
+
+type EventFlag uint8
+
+const (
+	EventFlagFill  EventFlag = 0x1
+	EventFlagOut   EventFlag = 0x2
+	EventFlagBid   EventFlag = 0x4
+	EventFlagMaker EventFlag = 0x8
+)
+
+type EventSide string
+
+const (
+	EventSideAsk EventSide = "ASK"
+	EventSideBid EventSide = "BID"
+)
+
+type Event struct {
+	Flag              EventFlag
+	OwnerSlot         uint8
+	FeeTier           uint8
+	Padding           [5]uint8
+	NativeQtyReleased uint64
+	NativeQtyPaid     uint64
+	NativeFeeOrRebate uint64
+	OrderID           bin.Uint128
+	Owner             solana.PublicKey
+	ClientOrderID     uint64
+}
+
+func (e *Event) Side() EventSide {
+	if Has(uint8(e.Flag), uint8(EventFlagBid)) {
+		return EventSideBid
+	}
+	return EventSideAsk
+}
+
+func (e *Event) Filled() bool {
+	return Has(uint8(e.Flag), uint8(EventFlagFill))
+}
+
+func Has(b, flag uint8) bool { return b&flag != 0 }

programs/serum/queue_test.go

@@ -0,0 +1,177 @@
+// Copyright 2020 dfuse Platform Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package serum
+
+import (
+	"context"
+	"encoding/base64"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/klauspost/compress/zstd"
+
+	"github.com/dfuse-io/solana-go"
+	"github.com/dfuse-io/solana-go/diff"
+	"github.com/dfuse-io/solana-go/rpc"
+
+	"github.com/stretchr/testify/assert"
+
+	bin "github.com/dfuse-io/binary"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestRequestQueue_Decoder(t *testing.T) {
+	bas64 := "c2VydW0JAAAAAAAAAAMAAAAAAAAAAAAAAAAAAACVlZMAAAAAAA0DBgEAAAAAECcAAAAAAABYvqFvAAAAAG1qbP//////TwcAAAAAAABbQ4ijQxgyr1dCuGPiALjnM85FHycAbicj2RmLNjNV0Y799DoWgU4WDQYGAQAAAAC4CwAAAAAAAMx8hiEAAAAAbGps//////9RBwAAAAAAAFtDiKNDGDKvV0K4Y+IAuOczzkUfJwBuJyPZGYs2M1XR6LcKOxaBThYNBwYBAAAAAIgTAAAAAAAAQILYNwAAAABramz//////1AHAAAAAAAAW0OIo0MYMq9XQrhj4gC45zPORR8nAG4nI9kZizYzVdE84P86FoFOFgYDAAAAAAAAjpWTAAAAAAAAAAAAAAAAAHtqbP//////TQcAAAAAAABbQ4ijQxgyr1dCuGPiALjnM85FHycAbicj2RmLNjNV0ZgF0ywMgU4WBgYAAAAAAACPlZMAAAAAAAAAAAAAAAAAemps//////9OBwAAAAAAAFtDiKNDGDKvV0K4Y+IAuOczzkUfJwBuJyPZGYs2M1XRkdfdLAyBThYGBwAAAAAAAJCVkwAAAAAAAAAAAAAAAACBamz//////08HAAAAAAAAW0OIo0MYMq9XQrhj4gC45zPORR8nAG4nI9kZizYzVdGFpMuDCYFOFgkABgEAAAAAuAsAAAAAAAAAAAAAAAAAAJGVkwAAAAAAWwcAAAAAAABbQ4ijQxgyr1dCuGPiALjnM85FHycAbicj2RmLNjNV0WFf6ToWgU4WAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAcGFkZGluZw=="
+	data, err := base64.StdEncoding.DecodeString(bas64)
+	require.NoError(t, err)
+
+	fmt.Println(hex.EncodeToString(data))
+
+	var q *RequestQueue
+	err = bin.NewDecoder(data).Decode(&q)
+	require.NoError(t, err)
+
+	assert.Equal(t, true, q.AccountFlags.Is(AccountFlagRequestQueue))
+	assert.Equal(t, true, q.AccountFlags.Is(AccountFlagInitialized))
+	assert.Equal(t, bin.Uint64(3), q.Head)
+	assert.Equal(t, bin.Uint64(9672085), q.NextSeqNum)
+	assert.Equal(t, bin.Uint64(0), q.Count)
+}
+
+func TestDecoder_EventQueue_Diff(t *testing.T) {
+	oldDataFile := "testdata/serum-event-queue-old.bin.zst"
+	newDataFile := "testdata/serum-event-queue-new.bin.zst"
+
+	// olDataJSONFile := strings.ReplaceAll(oldDataFile, ".bin.zst", ".json")
+	// newDataJSONFile := strings.ReplaceAll(newDataFile, ".bin.zst", ".json")
+
+	if os.Getenv("TESTDATA_UPDATE") == "true" {
+		client := rpc.NewClient("http://api.mainnet-beta.solana.com:80/rpc")
+		ctx := context.Background()
+		account := solana.MustPublicKeyFromBase58("13iGJcA4w5hcJZDjJbJQor1zUiDLE4jv2rMW9HkD5Eo1")
+
+		info, err := client.GetAccountInfo(ctx, account)
+		require.NoError(t, err)
+		writeCompressedFile(t, oldDataFile, info.Value.Data)
+
+		// oldQueue := &EventQueue{}
+		// require.NoError(t, oldQueue.Decode(info.Value.Data))
+		// writeJSONFile(t, olDataJSONFile, oldQueue)
+
+		time.Sleep(900 * time.Millisecond)
+
+		info, err = client.GetAccountInfo(ctx, account)
+		require.NoError(t, err)
+		writeCompressedFile(t, newDataFile, info.Value.Data)
+
+		// newQueue := &EventQueue{}
+		// require.NoError(t, newQueue.Decode(info.Value.Data))
+		// writeJSONFile(t, newDataJSONFile, newQueue)
+	}
+
+	oldQueue := &EventQueue{}
+	require.NoError(t, oldQueue.Decode(readCompressedFile(t, oldDataFile)))
+
+	newQueue := &EventQueue{}
+	require.NoError(t, newQueue.Decode(readCompressedFile(t, newDataFile)))
+
+	fmt.Println("==>> All diff(s)")
+	diff.Diff(oldQueue, newQueue, diff.OnEvent(func(event diff.Event) { fmt.Println("Event " + event.String()) }))
+}
+
+func TestDecoder_EventQueue_DiffManual(t *testing.T) {
+	oldQueue := &EventQueue{
+		SerumPadding: [5]byte{},
+		Head:         120,
+		Count:        13,
+		SeqNum:       25,
+		Events: []*Event{
+			{OrderID: bin.Uint128{Lo: 1}},
+			{OrderID: bin.Uint128{Lo: 2}},
+		},
+		EndPadding: [7]byte{},
+	}
+
+	newQueue := &EventQueue{
+		Head:   120,
+		Count:  13,
+		SeqNum: 25,
+		Events: []*Event{
+			{OrderID: bin.Uint128{Lo: 1}},
+			{OrderID: bin.Uint128{Lo: 4}},
+			{OrderID: bin.Uint128{Lo: 5}},
+		},
+	}
+
+	fmt.Println("All diff lines")
+	diff.Diff(oldQueue, newQueue, diff.OnEvent(func(event diff.Event) { fmt.Println("Event " + event.String()) }))
+
+	fmt.Println("")
+	fmt.Println("Processed diff lines")
+	diff.Diff(oldQueue, newQueue, diff.OnEvent(func(event diff.Event) {
+		if match, _ := event.Match("Events[#]"); match {
+			fmt.Printf("Event %s => %v\n", event.Kind, event.Element())
+		}
+	}))
+}
+
+func writeCompressedFile(t *testing.T, filename string, data []byte) {
+	require.NoError(t, ioutil.WriteFile(filename, zstEncoder.EncodeAll(data, nil), os.ModePerm), "unable to write compressed file %s", filename)
+}
+
+func readCompressedFile(t *testing.T, file string) []byte {
+	data, err := ioutil.ReadFile(file)
+	require.NoError(t, err)
+
+	out, err := zstDecoder.DecodeAll(data, nil)
+	require.NoError(t, err)
+
+	return out
+}
+
+var zstEncoder, _ = zstd.NewWriter(nil)
+var zstDecoder, _ = zstd.NewReader(nil)
+
+func writeFile(t *testing.T, filename string, data []byte) {
+	require.NoError(t, ioutil.WriteFile(filename, data, os.ModePerm), "unable to write file %s", filename)
+}
+
+func readFile(t *testing.T, file string) []byte {
+	data, err := ioutil.ReadFile(file)
+	require.NoError(t, err)
+
+	return data
+}
+
+func writeJSONFile(t *testing.T, filename string, v interface{}) {
+	out, err := json.MarshalIndent(v, "", "  ")
+	require.NoError(t, err)
+
+	require.NoError(t, ioutil.WriteFile(filename, out, os.ModePerm), "unable to write JSON file %s", filename)
+}
+
+func readJSONFile(t *testing.T, file string, v interface{}) {
+	data, err := ioutil.ReadFile(file)
+	require.NoError(t, err)
+
+	require.NoError(t, json.Unmarshal(data, v))
+	return
+}

programs/serum/rpc.go

@@ -21,8 +21,11 @@ import (
 	"fmt"
 
 	rice "github.com/GeertJohan/go.rice"
+	bin "github.com/dfuse-io/binary"
 	"github.com/dfuse-io/solana-go"
 	"github.com/dfuse-io/solana-go/rpc"
+	"github.com/dfuse-io/solana-go/rpc/ws"
+	"go.uber.org/zap"
 )
 
 //go:generate rice embed-go
@@ -95,3 +98,30 @@ func FetchMarket(ctx context.Context, rpcCli *rpc.Client, marketAddr solana.Publ
 
 	return meta, nil
 }
+
+func StreamOpenOrders(client *ws.Client) error {
+	sub, err := client.ProgramSubscribe(PROGRAM_ID, rpc.CommitmentSingleGossip)
+	if err != nil {
+		return fmt.Errorf("unable to subscribe to programID %q: %w", PROGRAM_ID, err)
+	}
+	count := 0
+	for {
+		d, err := sub.Recv()
+		if err != nil {
+			return fmt.Errorf("received error from programID subscription: %w", err)
+		}
+		res := d.(*ws.ProgramResult)
+
+		var f *AccountFlag
+		err = bin.NewDecoder(res.Value.Account.Data).Decode(&f)
+		if err != nil {
+			fmt.Println("***********************************", err)
+			zlog.Debug("unable to decoce account flag for account... skipping",
+				zap.Stringer("account_address", res.Value.PubKey),
+			)
+			continue
+		}
+		fmt.Printf("%d - %s\n", count, f.String())
+		count++
+	}
+}