6609350879

Committed 23 Oct 2023 06:05AM UTC coverage: 5.771% (+1.0%) from 4.739%

Build Type

github

Committed by Viktor Nosov

Commit Message

mod

Pull Request Pull Request #136: services refactoring

Run Details

902 of 902 new or added lines in 33 files covered. (100.0%)

468 of 8109 relevant lines covered (5.77%)

0.26 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/client/chaincode/txwaiter/all.go

package txwaiter

import (
        "context"
        "sync"

        "github.com/pkg/errors"

        "github.com/s7techlab/hlf-sdk-go/api"
        clienterr "github.com/s7techlab/hlf-sdk-go/client/errors"
)

// All - need use on invoke flow for check transaction codes for each organization from endorsement policy
// txwaiter.All  will be made to subscribe tx for each of the peer organizations from the endorsement policy
func All(cfg *api.DoOptions) (api.TxWaiter, error) {
        waiter := &allMspWaiter{
                onceSet: new(sync.Once),
        }

        // make delivers for each mspID
        errD := new(clienterr.MultiError)
        for i := range cfg.EndorsingMspIDs {
                peerDeliver, err := cfg.Pool.DeliverClient(cfg.EndorsingMspIDs[i], cfg.Identity)
                if err != nil {
                        errD.Add(errors.Wrapf(err, "%s: failed to get delivery client", cfg.EndorsingMspIDs[i]))
                        continue
                }

                waiter.delivers = append(waiter.delivers, peerDeliver)
        }
        if len(errD.Errors) != 0 {
                return nil, errD
        }

        return waiter, nil
}

type allMspWaiter struct {
        delivers []api.DeliverClient
        onceSet  *sync.Once
        hasErr   bool
}

func (w *allMspWaiter) setErr() {
        w.onceSet.Do(func() { w.hasErr = true })
}

// Wait - implementation of api.TxWaiter interface
func (w *allMspWaiter) Wait(ctx context.Context, channel string, txId string) error {
        var (
                wg   = new(sync.WaitGroup)
                errS = make(chan error, len(w.delivers))
        )

        for i := range w.delivers {
                wg.Add(1)
                go func(j int) {
                        err := waitPerOne(ctx, w.delivers[j], channel, txId)
                        if err != nil {
                                w.setErr()
                                errS <- err
                        }
                        wg.Done()
                }(i)
        }
        wg.Wait()
        close(errS)

        if w.hasErr {
                mErr := &clienterr.MultiError{}
                for e := range errS {
                        if e != nil {
                                mErr.Errors = append(mErr.Errors, e)
                        }
                }
                return mErr
        }

        return nil
}

func waitPerOne(ctx context.Context, deliver api.DeliverClient, channelName string, txId string) error {
        sub, err := deliver.SubscribeTx(ctx, channelName, txId)
        if err != nil {
                return errors.Wrap(err, "failed to subscribe on tx event")
        }
        defer func() { _ = sub.Close() }()

        _, err = sub.Result()
        return err
}

1	package txwaiter
2
3	import (
4	"context"
5	"sync"
6
7	"github.com/pkg/errors"
8
9	"github.com/s7techlab/hlf-sdk-go/api"
10	clienterr "github.com/s7techlab/hlf-sdk-go/client/errors"
11	)
12
13	// All - need use on invoke flow for check transaction codes for each organization from endorsement policy
14	// txwaiter.All will be made to subscribe tx for each of the peer organizations from the endorsement policy
15	func All(cfg *api.DoOptions) (api.TxWaiter, error) {	×
16	waiter := &allMspWaiter{	×
17	onceSet: new(sync.Once),	×
18	}	×
19		×
20	// make delivers for each mspID	×
21	errD := new(clienterr.MultiError)	×
22	for i := range cfg.EndorsingMspIDs {	×
23	peerDeliver, err := cfg.Pool.DeliverClient(cfg.EndorsingMspIDs[i], cfg.Identity)	×
24	if err != nil {	×
25	errD.Add(errors.Wrapf(err, "%s: failed to get delivery client", cfg.EndorsingMspIDs[i]))	×
26	continue	×
27	}
28
29	waiter.delivers = append(waiter.delivers, peerDeliver)	×
30	}
31	if len(errD.Errors) != 0 {	×
32	return nil, errD	×
33	}	×
34
35	return waiter, nil	×
36	}
37
38	type allMspWaiter struct {
39	delivers []api.DeliverClient
40	onceSet *sync.Once
41	hasErr bool
42	}
43
44	func (w *allMspWaiter) setErr() {	×
45	w.onceSet.Do(func() { w.hasErr = true })	×
46	}
47
48	// Wait - implementation of api.TxWaiter interface
49	func (w *allMspWaiter) Wait(ctx context.Context, channel string, txId string) error {	×
50	var (	×
51	wg = new(sync.WaitGroup)	×
52	errS = make(chan error, len(w.delivers))	×
53	)	×
54		×
55	for i := range w.delivers {	×
56	wg.Add(1)	×
57	go func(j int) {	×
58	err := waitPerOne(ctx, w.delivers[j], channel, txId)	×
59	if err != nil {	×
60	w.setErr()	×
61	errS <- err	×
62	}	×
63	wg.Done()	×
64	}(i)
65	}
66	wg.Wait()	×
67	close(errS)	×
68		×
69	if w.hasErr {	×
70	mErr := &clienterr.MultiError{}	×
71	for e := range errS {	×
72	if e != nil {	×
73	mErr.Errors = append(mErr.Errors, e)	×
74	}	×
75	}
76	return mErr	×
77	}
78
79	return nil	×
80	}
81
82	func waitPerOne(ctx context.Context, deliver api.DeliverClient, channelName string, txId string) error {	×
83	sub, err := deliver.SubscribeTx(ctx, channelName, txId)	×
84	if err != nil {	×
85	return errors.Wrap(err, "failed to subscribe on tx event")	×
86	}	×
87	defer func() { _ = sub.Close() }()	×
88
89	_, err = sub.Result()	×
90	return err	×
91	}

s7techlab / hlf-sdk-go / 6609350879

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous