Skip to content

Commit cc8e99d

Browse files
jsignjsign
committed
lotus: decoupled lotus client #1
Signed-off-by: jsign <jsign.uy@gmail.com>
1 parent aaf0d9e commit cc8e99d

54 files changed

Lines changed: 2313 additions & 7960 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎.circleci/config.yml‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
version: 2.1
2+
3+
jobs:
4+
test:
5+
docker:
6+
- image: circleci/golang:1.13
7+
steps:
8+
- checkout
9+
- restore_cache:
10+
key: go-mod-v1-{{ checksum "go.sum" }}-{{ arch }}
11+
- run:
12+
name: cache mods
13+
command: |
14+
go mod download
15+
- save_cache:
16+
key: go-mod-v1-{{ checksum "go.sum" }}-{{ arch }}
17+
paths:
18+
- /go/pkg/mod
19+
20+
workflows:
21+
version: 2
22+
filecoin:
23+
jobs:
24+
- test:
25+
filters:
26+
tags:
27+
only: /.*/

‎client/client.go‎

Lines changed: 0 additions & 20 deletions
This file was deleted.

‎client/client_test.go‎

Lines changed: 0 additions & 40 deletions
This file was deleted.

‎deals/askcache.go‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -3,12 +3,13 @@ package deals
33
import (
44
"bytes"
55
"context"
6+
"encoding/json"
67
"sort"
78
"sync"
89
"time"
910

10-
"github.com/filecoin-project/lotus/chain/types"
1111
"github.com/ipfs/go-datastore"
12+
"github.com/textileio/filecoin/lotus/types"
1213
)
1314

1415
var (
@@ -54,7 +55,7 @@ func (d *DealModule) AvailableAsks(q Query) ([]StorageAsk, error) {
5455
res = append(res, StorageAsk{
5556
Price: sa.Price.Uint64(),
5657
MinPieceSize: sa.MinPieceSize,
57-
Miner: sa.Miner.String(),
58+
Miner: sa.Miner,
5859
Timestamp: sa.Timestamp,
5960
Expiry: sa.Expiry,
6061
})
@@ -94,13 +95,14 @@ func (d *DealModule) updateMinerAsks() error {
9495
})
9596

9697
var buf bytes.Buffer
98+
encoder := json.NewEncoder(&buf)
9799
for _, ask := range asks {
98100
buf.Reset()
99-
if err := ask.MarshalCBOR(&buf); err != nil {
101+
if err := encoder.Encode(ask); err != nil {
100102
log.Errorf("error when marshaling storage ask: %s", err)
101103
return err
102104
}
103-
if err := d.ds.Put(dsStorageAskBase.ChildString(ask.Miner.String()), buf.Bytes()); err != nil {
105+
if err := d.ds.Put(dsStorageAskBase.ChildString(ask.Miner), buf.Bytes()); err != nil {
104106
log.Errorf("error when persiting storage ask: %s", err)
105107
return err
106108
}

‎deals/deals.go‎

Lines changed: 27 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -6,18 +6,16 @@ import (
66
"io"
77
"io/ioutil"
88
"os"
9+
"path/filepath"
910
"reflect"
1011
"sync"
1112
"time"
1213

13-
"github.com/filecoin-project/lotus/api"
14-
"github.com/filecoin-project/lotus/chain/address"
15-
"github.com/filecoin-project/lotus/chain/store"
16-
"github.com/filecoin-project/lotus/chain/types"
1714
"github.com/ipfs/go-cid"
1815
"github.com/ipfs/go-datastore"
1916
logging "github.com/ipfs/go-log"
2017
peer "github.com/libp2p/go-libp2p-core/peer"
18+
"github.com/textileio/filecoin/lotus/types"
2119
)
2220

2321
const (
@@ -34,19 +32,20 @@ var (
3432

3533
// DealerAPI interacts with a Filecoin full-node
3634
type DealerAPI interface {
37-
ClientStartDeal(ctx context.Context, data cid.Cid, addr address.Address, miner address.Address, epochPrice types.BigInt, blocksDuration uint64) (*cid.Cid, error)
35+
ClientStartDeal(ctx context.Context, data cid.Cid, addr string, miner string, epochPrice types.BigInt, blocksDuration uint64) (*cid.Cid, error)
3836
ClientImport(ctx context.Context, path string) (cid.Cid, error)
39-
ClientGetDealInfo(context.Context, cid.Cid) (*api.DealInfo, error)
40-
ChainNotify(context.Context) (<-chan []*store.HeadChange, error)
41-
StateListMiners(context.Context, *types.TipSet) ([]address.Address, error)
42-
ClientQueryAsk(ctx context.Context, p peer.ID, miner address.Address) (*types.SignedStorageAsk, error)
43-
StateMinerPeerID(ctx context.Context, m address.Address, ts *types.TipSet) (peer.ID, error)
37+
ClientGetDealInfo(context.Context, cid.Cid) (*types.DealInfo, error)
38+
ChainNotify(context.Context) (<-chan struct{}, error)
39+
StateListMiners(context.Context, *types.TipSet) ([]string, error)
40+
ClientQueryAsk(ctx context.Context, p peer.ID, miner string) (*types.SignedStorageAsk, error)
41+
StateMinerPeerID(ctx context.Context, m string, ts *types.TipSet) (peer.ID, error)
4442
}
4543

4644
// DealModule exposes storage, monitoring, and Asks from the market.
4745
type DealModule struct {
48-
api DealerAPI
49-
ds datastore.Datastore
46+
api DealerAPI
47+
ds datastore.Datastore
48+
basePathImport string
5049

5150
askCacheLock sync.RWMutex
5251
askCache []*types.StorageAsk
@@ -79,11 +78,17 @@ type DealInfo struct {
7978

8079
// New creates a new deal module
8180
func New(api DealerAPI, ds datastore.Datastore) *DealModule {
81+
// can't avoid home base path, ipfs checks: cannot add filestore references outside ipfs root (home folder)
82+
home, err := os.UserHomeDir()
83+
if err != nil {
84+
panic(err)
85+
}
8286
dm := &DealModule{
83-
api: api,
84-
ds: ds,
85-
close: make(chan struct{}),
86-
closed: make(chan struct{}),
87+
api: api,
88+
ds: ds,
89+
basePathImport: filepath.Join(home, "textilefc"),
90+
close: make(chan struct{}),
91+
closed: make(chan struct{}),
8792
}
8893
go dm.runBackgroundAskCache()
8994
return dm
@@ -104,7 +109,7 @@ func (d *DealModule) Close() {
104109
// Store creates a proposal deal for data using wallet addr to all miners indicated
105110
// by dealConfigs for duration epochs
106111
func (d *DealModule) Store(ctx context.Context, addr string, data io.Reader, dealConfigs []DealConfig, duration uint64) ([]cid.Cid, []DealConfig, error) {
107-
tmpF, err := ioutil.TempFile("", "import-*")
112+
tmpF, err := ioutil.TempFile(d.basePathImport, "import-*")
108113
if err != nil {
109114
return nil, nil, fmt.Errorf("error when creating tmpfile: %s", err)
110115
}
@@ -118,20 +123,15 @@ func (d *DealModule) Store(ctx context.Context, addr string, data io.Reader, dea
118123
return nil, nil, fmt.Errorf("error when importing data: %s", err)
119124
}
120125

121-
myAddr, err := address.NewFromString(addr)
122-
if err != nil {
123-
return nil, nil, fmt.Errorf("wallet addr is invalid: %s", err)
124-
}
125126
var proposals []cid.Cid
126127
var failed []DealConfig
127128
for _, dconfig := range dealConfigs {
128-
minerAddr, err := address.NewFromString(dconfig.Miner)
129129
if err != nil {
130130
log.Errorf("miner addr is invalid %v: %s", dconfig, err)
131131
failed = append(failed, dconfig)
132132
continue
133133
}
134-
proposal, err := d.api.ClientStartDeal(ctx, dataCid, myAddr, minerAddr, dconfig.EpochPrice, duration)
134+
proposal, err := d.api.ClientStartDeal(ctx, dataCid, addr, dconfig.Miner, dconfig.EpochPrice, duration)
135135
if err != nil {
136136
log.Errorf("error when starting deal with %v: %s", dconfig, err)
137137
failed = append(failed, dconfig)
@@ -152,7 +152,7 @@ func (d *DealModule) Watch(ctx context.Context, proposals []cid.Cid) (<-chan Dea
152152
go func() {
153153
defer close(ch)
154154

155-
currentState := make(map[cid.Cid]api.DealInfo)
155+
currentState := make(map[cid.Cid]types.DealInfo)
156156
tout := time.After(initialWait)
157157
for {
158158
select {
@@ -172,7 +172,7 @@ func (d *DealModule) Watch(ctx context.Context, proposals []cid.Cid) (<-chan Dea
172172
return ch, nil
173173
}
174174

175-
func (d *DealModule) pushNewChanges(ctx context.Context, currState map[cid.Cid]api.DealInfo, proposals []cid.Cid, ch chan<- DealInfo) error {
175+
func (d *DealModule) pushNewChanges(ctx context.Context, currState map[cid.Cid]types.DealInfo, proposals []cid.Cid, ch chan<- DealInfo) error {
176176
for _, pcid := range proposals {
177177
dinfo, err := d.api.ClientGetDealInfo(ctx, pcid)
178178
if err != nil {
@@ -184,8 +184,8 @@ func (d *DealModule) pushNewChanges(ctx context.Context, currState map[cid.Cid]a
184184
newState := DealInfo{
185185
ProposalCid: dinfo.ProposalCid,
186186
StateID: dinfo.State,
187-
StateName: api.DealStates[dinfo.State],
188-
Miner: dinfo.Provider.String(),
187+
StateName: types.DealStates[dinfo.State],
188+
Miner: dinfo.Provider,
189189
PieceRef: dinfo.PieceRef,
190190
Size: dinfo.Size,
191191
PricePerEpoch: dinfo.PricePerEpoch,

‎deals/deals_test.go‎

Lines changed: 8 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -2,35 +2,20 @@ package deals
22

33
import (
44
"fmt"
5-
"os"
65
"reflect"
76
"testing"
87

9-
"github.com/filecoin-project/lotus/chain/address"
10-
"github.com/filecoin-project/lotus/chain/types"
11-
"github.com/textileio/filecoin/client"
8+
"github.com/textileio/filecoin/lotus"
9+
"github.com/textileio/filecoin/lotus/types"
1210
"github.com/textileio/filecoin/tests"
1311
)
1412

15-
var (
16-
authToken = ""
17-
)
18-
19-
func TestMain(m *testing.M) {
20-
var err error
21-
authToken, err = tests.GetLotusToken()
22-
if err != nil {
23-
fmt.Println("couldn't get/generate lotus authtoken")
24-
os.Exit(-1)
25-
}
26-
os.Exit(m.Run())
27-
}
28-
2913
func TestAskCache(t *testing.T) {
3014
if testing.Short() {
3115
t.Skip("skipping test since we're on short mode")
3216
}
33-
c, cls, err := client.New(tests.DaemonAddr, authToken)
17+
addr, token := tests.ClientConfig(t)
18+
c, cls, err := lotus.New(addr, token)
3419
checkErr(t, err)
3520
defer cls()
3621

@@ -47,10 +32,10 @@ func TestQueryAsk(t *testing.T) {
4732
t.Parallel()
4833
dm := DealModule{}
4934
dm.askCache = []*types.StorageAsk{
50-
{Price: types.NewInt(20), MinPieceSize: 128, Miner: newaddr("t01")},
51-
{Price: types.NewInt(30), MinPieceSize: 64, Miner: newaddr("t02")},
52-
{Price: types.NewInt(40), MinPieceSize: 256, Miner: newaddr("t03")},
53-
{Price: types.NewInt(50), MinPieceSize: 16, Miner: newaddr("t04")},
35+
{Price: types.NewInt(20), MinPieceSize: 128, Miner: "t01"},
36+
{Price: types.NewInt(30), MinPieceSize: 64, Miner: "t02"},
37+
{Price: types.NewInt(40), MinPieceSize: 256, Miner: "t03"},
38+
{Price: types.NewInt(50), MinPieceSize: 16, Miner: "t04"},
5439
}
5540

5641
facr := []StorageAsk{
@@ -87,14 +72,6 @@ func TestQueryAsk(t *testing.T) {
8772
}
8873
}
8974

90-
func newaddr(s string) address.Address {
91-
a, err := address.NewFromString(s)
92-
if err != nil {
93-
panic("invalid address")
94-
}
95-
return a
96-
}
97-
9875
func checkErr(t *testing.T, err error) {
9976
t.Helper()
10077
if err != nil {

0 commit comments

Comments
 (0)