Merge pull request #1546 from 0chain/fix/wasm-ready

Wait for start event from web workers

wasmsdk/blobber.go

@@ -19,6 +19,7 @@ import (
 	"github.com/0chain/gosdk/core/common"
 	"github.com/0chain/gosdk/core/pathutil"
 	"github.com/0chain/gosdk/core/sys"
+	"github.com/hack-pad/safejs"
 
 	"github.com/0chain/gosdk/core/transaction"
 	"github.com/0chain/gosdk/wasmsdk/jsbridge"
@@ -629,6 +630,12 @@ func multiUpload(jsonBulkUploadOptions string) (MultiUploadResult, error) {
 		result.Success = false
 		return result, errors.New("Error fetching the allocation")
 	}
+	err = addWebWorkers(allocationObj)
+	if err != nil {
+		result.Error = err.Error()
+		result.Success = false
+		return result, err
+	}
 
 	operationRequests := make([]sdk.OperationRequest, n)
 	for idx, option := range options {
@@ -1019,10 +1026,10 @@ func terminateWorkersWithAllocation(alloc *sdk.Allocation) {
 	}
 }
 
-func createWorkers(allocationID string) bool {
+func createWorkers(allocationID string) error {
 	alloc, err := getAllocation(allocationID)
 	if err != nil {
-		return false
+		return err
 	}
 	return addWebWorkers(alloc)
 }
@@ -1035,6 +1042,8 @@ func startListener() error {
 	if err != nil {
 		return err
 	}
+	safeVal, _ := safejs.ValueOf("startListener")
+	selfWorker.PostMessage(safeVal, nil) //nolint:errcheck
 
 	listener, err := selfWorker.Listen(ctx)
 	if err != nil {

wasmsdk/cache.go

@@ -4,6 +4,8 @@
 package main
 
 import (
+	"context"
+	"errors"
 	"time"
 
 	"github.com/0chain/gosdk/wasmsdk/jsbridge"
@@ -66,17 +68,52 @@ func reloadAllocation(allocationID string) (*sdk.Allocation, error) {
 	return it.Allocation, nil
 }
 
-func addWebWorkers(alloc *sdk.Allocation) (isCreated bool) {
+func addWebWorkers(alloc *sdk.Allocation) (err error) {
 	c := client.GetClient()
 	if c == nil || len(c.Keys) == 0 {
 		return
 	}
-
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+	defer cancel()
+	respChan := make(chan error, len(alloc.Blobbers))
+	respRequired := 0
 	for _, blober := range alloc.Blobbers {
-		_, workerCreated, _ := jsbridge.NewWasmWebWorker(blober.ID, blober.Baseurl, c.ClientID, c.Keys[0].PublicKey, c.Keys[0].PrivateKey, c.Mnemonic) //nolint:errcheck
+		weborker, workerCreated, _ := jsbridge.NewWasmWebWorker(blober.ID, blober.Baseurl, c.ClientID, c.Keys[0].PublicKey, c.Keys[0].PrivateKey, c.Mnemonic) //nolint:errcheck
 		if workerCreated {
-			isCreated = true
+			respRequired++
+			go func() {
+				eventChan, err := weborker.Listen(ctx)
+				if err != nil {
+					respChan <- err
+					return
+				}
+				_, ok := <-eventChan
+				if !ok {
+					respChan <- errors.New("worker chan closed")
+					return
+				}
+				respChan <- nil
+			}()
+		}
+	}
+	if respRequired == 0 {
+		return
+	}
+	for {
+		select {
+		case <-ctx.Done():
+			PrintError(ctx.Err())
+			return ctx.Err()
+		case err = <-respChan:
+			if err != nil {
+				PrintError(err)
+				return
+			}
+			respRequired--
+			if respRequired == 0 {
+				close(respChan)
+				return
+			}
 		}
 	}
-	return
 }

wasmsdk/jsbridge/webworker.go

@@ -8,6 +8,7 @@ import (
 	"errors"
 	"sync"
 
+	"github.com/0chain/gosdk/core/version"
 	"github.com/google/uuid"
 	"github.com/hack-pad/go-webworkers/worker"
 	"github.com/hack-pad/safejs"
@@ -60,7 +61,7 @@ func NewWasmWebWorker(blobberID, blobberURL, clientID, publicKey, privateKey, mn
 	w := &WasmWebWorker{
 		Name:        blobberURL,
 		Env:         []string{"BLOBBER_URL=" + blobberURL, "CLIENT_ID=" + clientID, "PRIVATE_KEY=" + privateKey, "MODE=worker", "PUBLIC_KEY=" + publicKey, "MNEMONIC=" + mnemonic},
-		Path:        "zcn.wasm",
+		Path:        "zcn.wasm?v=" + version.VERSIONSTR,
 		subscribers: make(map[string]chan worker.MessageEvent),
 	}
 
@@ -108,6 +109,7 @@ func (ww *WasmWebWorker) SubscribeToEvents(remotePath string, ch chan worker.Mes
 		ww.cancelContext = cancel
 		eventChan, err := ww.Listen(ctx)
 		if err != nil {
+			ww.subMutex.Unlock()
 			return err
 		}
 		go ww.ListenForEvents(eventChan)