timeout and stats

core/schemaclient.go

@@ -19,12 +19,18 @@ type SchemaClientOpts struct {
 	SetNode    func(pos *mt.Pos, node *mt.Node, md *parser.MetadataEntry) error
 }
 
+type SchemaClientStats struct {
+	UsedMapblocks     int
+	ReceivedMapblocks int
+}
+
 type SchemaClient struct {
 	opts            *SchemaClientOpts
 	origin          *mt.Pos
 	area            *mt.Area
 	ser_ver         uint8
 	id_node_mapping map[int]string
+	stats           *SchemaClientStats
 }
 
 func NewSchemaClient(opts *SchemaClientOpts) *SchemaClient {
@@ -44,9 +50,14 @@ func NewSchemaClient(opts *SchemaClientOpts) *SchemaClient {
 			CONTENT_IGNORE:  "ignore",
 			CONTENT_UNKNOWN: "unknown",
 		},
+		stats: &SchemaClientStats{},
 	}
 }
 
+func (sc *SchemaClient) Stats() *SchemaClientStats {
+	return sc.stats
+}
+
 func (sc *SchemaClient) applyBlockChanges(mb_startpos *mt.Pos, mb *mt.MapBlock) error {
 
 	for i := 0; i < 4096; i++ {
@@ -104,14 +115,15 @@ func (sc *SchemaClient) serverInitHandler(client *commandclient.CommandClient) {
 	}
 }
 
-func (sc *SchemaClient) blockDataHandler(client *commandclient.CommandClient, errchan chan error) {
+func (sc *SchemaClient) blockDataHandler(client *commandclient.CommandClient) error {
 	for o := range client.CommandChannel() {
 		switch cmd := o.(type) {
 		case *commands.ServerBlockData:
 			pos1 := cmd.Pos.Multiply(16)
 			pos2 := pos1.Add(mt.NewPos(15, 15, 15))
 			area := mt.NewArea(pos1, pos2)
 			fmt.Printf("Blockdata: %d bytes, pos1: %s, pos2: %s\n", len(cmd.BlockData), pos1, pos2)
+			sc.stats.ReceivedMapblocks++
 
 			if !area.Intersects(sc.area) {
 				// no valid data
@@ -120,29 +132,36 @@ func (sc *SchemaClient) blockDataHandler(client *commandclient.CommandClient, er
 
 			mb, err := mapparser.ParseNetwork(sc.ser_ver, cmd.BlockData)
 			if err != nil {
-				errchan <- fmt.Errorf("map parse error @ %v: %v", cmd.Pos, err)
-				return
+				return fmt.Errorf("map parse error @ %v: %v", cmd.Pos, err)
 			}
 
 			err = sc.applyBlockChanges(pos1, mb)
 			if err != nil {
-				errchan <- fmt.Errorf("apply block changes error: %v", err)
-				return
+				return fmt.Errorf("apply block changes error: %v", err)
 			}
+
+			sc.stats.UsedMapblocks++
 		}
 	}
+
+	return nil
 }
 
-func (sc *SchemaClient) Run() error {
+func (sc *SchemaClient) Run(d time.Duration) error {
 
 	client := commandclient.NewCommandClient(sc.opts.Pull.Hostname, sc.opts.Pull.Port)
-	errchan := make(chan error)
 
 	err := client.Connect()
 	if err != nil {
 		return fmt.Errorf("connect error: %v", err)
 	}
 
+	go func() {
+		// disconnect after timeout
+		time.Sleep(d)
+		client.Disconnect()
+	}()
+
 	err = commandclient.Init(client, sc.opts.PullClient.Username)
 	if err != nil {
 		return fmt.Errorf("init error: %v", err)
@@ -153,26 +172,16 @@ func (sc *SchemaClient) Run() error {
 		return fmt.Errorf("login error: %v", err)
 	}
 
-	// TODO: timeout
 	sc.serverInitHandler(client)
 
 	err = clientReady(client)
 	if err != nil {
 		return fmt.Errorf("clientready error: %v", err)
 	}
 
-	go sc.blockDataHandler(client, errchan)
-
-	select {
-	case <-time.After(5 * time.Second):
-		break
-	case err = <-errchan:
-		return fmt.Errorf("blockhandler error: %v", err)
-	}
-
-	err = client.Disconnect()
+	err = sc.blockDataHandler(client)
 	if err != nil {
-		return fmt.Errorf("disconnect error: %v", err)
+		return fmt.Errorf("blockhandler error: %v", err)
 	}
 
 	return nil

core/schemaclient_test.go

@@ -6,6 +6,7 @@ import (
 	"blockexchange/types"
 	"fmt"
 	"testing"
+	"time"
 
 	mt "github.com/minetest-go/types"
 
@@ -46,8 +47,9 @@ func TestSchemaClient(t *testing.T) {
 	}
 
 	sc := core.NewSchemaClient(opts)
-	err := sc.Run()
+	err := sc.Run(time.Second * 5)
 	assert.NoError(t, err)
+	fmt.Printf("Stats: %v\n", sc.Stats())
 	t.FailNow()
 
 }