cleanup Swarm.Peers
This commit was moved from ipfs/go-ipfs-http-client@fd7858dc57b86761599b0e844f3536f2d4a0d953
Łukasz Magiera committed
Feb 19, 2019 at 23:11 UTC
f34788e6ee61aad0ee1970a4f3d004bdebdbcd27
7 files changed
+69
-66
client/httpapi/api.go
-1
@@ -124,7 +124,6 @@ func (api *HttpApi) request(command string, args ...string) *RequestBuilder {
124
command: command,
125
args: args,
126
shell: api,
127
- drainOut: true,
127
}
128
}
129
client/httpapi/apifile.go
+7
-7
@@ -49,15 +49,15 @@ type apiFile struct {
49
size int64
50
path iface.Path
51
52
- r io.ReadCloser
52
+ r *Response
53
at int64
54
}
55
56
func (f *apiFile) reset() error {
57
if f.r != nil {
58
- f.r.Close()
58
+ f.r.Cancel()
59
}
60
- req := f.core.request("cat", f.path.String()).NoDrain()
60
+ req := f.core.request("cat", f.path.String())
61
if f.at != 0 {
62
req.Option("offset", f.at)
63
}
@@ -68,12 +68,12 @@ func (f *apiFile) reset() error {
68
if resp.Error != nil {
69
return resp.Error
70
}
71
- f.r = resp.Output
71
+ f.r = resp
72
return nil
73
}
74
75
func (f *apiFile) Read(p []byte) (int, error) {
76
- n, err := f.r.Read(p)
76
+ n, err := f.r.Output.Read(p)
77
if n > 0 {
78
f.at += int64(n)
79
}
@@ -92,7 +92,7 @@ func (f *apiFile) Seek(offset int64, whence int) (int64, error) {
92
}
93
94
if f.at < offset && offset-f.at < forwardSeekLimit { //forward skip
95
- r, err := io.CopyN(ioutil.Discard, f.r, offset-f.at)
95
+ r, err := io.CopyN(ioutil.Discard, f.r.Output, offset-f.at)
96
97
f.at += r
98
return f.at, err
@@ -103,7 +103,7 @@ func (f *apiFile) Seek(offset int64, whence int) (int64, error) {
103
104
func (f *apiFile) Close() error {
105
if f.r != nil {
106
- return f.r.Close()
106
+ return f.r.Cancel()
107
}
108
return nil
109
}
client/httpapi/pubsub.go
+6
-3
@@ -60,7 +60,7 @@ type pubsubSub struct {
60
messages chan pubsubMessage
61
62
done chan struct{}
63
- rcloser io.Closer
63
+ rcloser func() error
64
}
65
66
type pubsubMessage struct {
@@ -113,7 +113,7 @@ func (api *PubsubAPI) Subscribe(ctx context.Context, topic string, opts ...caopt
113
}
114
115
resp, err := api.core().request("pubsub/sub", topic).
116
- Option("discover", options.Discover).NoDrain().Send(ctx)
116
+ Option("discover", options.Discover).Send(ctx)
117
118
if err != nil {
119
return nil, err
@@ -125,6 +125,9 @@ func (api *PubsubAPI) Subscribe(ctx context.Context, topic string, opts ...caopt
125
sub := &pubsubSub{
126
messages: make(chan pubsubMessage),
127
done: make(chan struct{}),
128
+ rcloser: func() error {
129
+ return resp.Cancel()
130
+ },
131
}
132
133
dec := json.NewDecoder(resp.Output)
@@ -159,7 +162,7 @@ func (s *pubsubSub) Close() error {
162
close(s.done)
163
s.done = nil
164
}
162
- return s.rcloser.Close()
165
+ return s.rcloser()
166
}
167
168
func (api *PubsubAPI) core() *HttpApi {
client/httpapi/request.go
-2
@@ -13,7 +13,6 @@ type Request struct {
13
Opts map[string]string
14
Body io.Reader
15
Headers map[string]string
16
- DrainOut bool // if set, resp.Close will read all remaining data
16
}
17
18
func NewRequest(ctx context.Context, url, command string, args ...string) *Request {
@@ -31,6 +30,5 @@ func NewRequest(ctx context.Context, url, command string, args ...string) *Reque
30
Args: args,
31
Opts: opts,
32
Headers: make(map[string]string),
34
- DrainOut: true,
33
}
34
}
client/httpapi/requestbuilder.go
-7
@@ -19,7 +19,6 @@ type RequestBuilder struct {
19
opts map[string]string
20
headers map[string]string
21
body io.Reader
22
- drainOut bool
22
23
shell *HttpApi
24
}
@@ -85,12 +84,6 @@ func (r *RequestBuilder) Header(name, value string) *RequestBuilder {
84
return r
85
}
86
88
-// NoDrain disables output draining in response closer
89
-func (r *RequestBuilder) NoDrain() *RequestBuilder {
90
- r.drainOut = false
91
- return r
92
-}
93
-
87
// Send sends the request and return the response.
88
func (r *RequestBuilder) Send(ctx context.Context) (*Response, error) {
89
r.shell.applyGlobal(r)
client/httpapi/response.go
+11
-8
@@ -34,18 +34,13 @@ func (r *trailerReader) Close() error {
34
type Response struct {
35
Output io.ReadCloser
36
Error *Error
37
-
38
- drainOutput bool
37
}
38
39
func (r *Response) Close() error {
40
if r.Output != nil {
41
44
- // always drain output (response body)
45
- var err1 error
46
- if r.drainOutput {
47
- _, err1 = io.Copy(ioutil.Discard, r.Output)
48
- }
42
+ // drain output (response body)
43
+ _, err1 := io.Copy(ioutil.Discard, r.Output)
44
err2 := r.Output.Close()
45
if err1 != nil {
46
return err1
@@ -55,6 +50,15 @@ func (r *Response) Close() error {
50
return nil
51
}
52
53
+// Cancel aborts running request (without draining request body)
54
+func (r *Response) Cancel() error {
55
+ if r.Output != nil {
56
+ return r.Output.Close()
57
+ }
58
+
59
+ return nil
60
+}
61
+
62
func (r *Response) Decode(dec interface{}) error {
63
defer r.Close()
64
if r.Error != nil {
@@ -123,7 +127,6 @@ func (r *Request) Send(c *http.Client) (*Response, error) {
127
128
nresp := new(Response)
129
126
- nresp.drainOutput = r.DrainOut
130
nresp.Output = &trailerReader{resp}
131
if resp.StatusCode >= http.StatusBadRequest {
132
e := &Error{
client/httpapi/swarm.go
+45
-38
@@ -32,74 +32,81 @@ func (api *SwarmAPI) Disconnect(ctx context.Context, addr multiaddr.Multiaddr) e
32
return api.core().request("swarm/disconnect", addr.String()).Exec(ctx, nil)
33
}
34
35
-type streamInfo struct {
36
- Protocol string
37
-}
38
-
35
type connInfo struct {
40
- Addr string
41
- Peer string
42
- JLatency time.Duration `json:"Latency"`
43
- Muxer string
44
- JDirection inet.Direction `json:"Direction"`
45
- JStreams []streamInfo `json:"Streams"`
46
-}
47
-
48
-func (c *connInfo) valid() error {
49
- _, err := multiaddr.NewMultiaddr(c.Addr)
50
- if err != nil {
51
- return err
52
- }
53
-
54
- _, err = peer.IDB58Decode(c.Peer)
55
- return err
36
+ addr multiaddr.Multiaddr
37
+ peer peer.ID
38
+ latency time.Duration
39
+ muxer string
40
+ direction inet.Direction
41
+ streams []protocol.ID
42
}
43
44
func (c *connInfo) ID() peer.ID {
59
- id, _ := peer.IDB58Decode(c.Peer)
60
- return id
45
+ return c.peer
46
}
47
48
func (c *connInfo) Address() multiaddr.Multiaddr {
64
- a, _ := multiaddr.NewMultiaddr(c.Addr)
65
- return a
49
+ return c.addr
50
}
51
52
func (c *connInfo) Direction() inet.Direction {
69
- return c.JDirection
53
+ return c.direction
54
}
55
56
func (c *connInfo) Latency() (time.Duration, error) {
73
- return c.JLatency, nil
57
+ return c.latency, nil
58
}
59
60
func (c *connInfo) Streams() ([]protocol.ID, error) {
77
- res := make([]protocol.ID, len(c.JStreams))
78
- for i, stream := range c.JStreams {
79
- res[i] = protocol.ID(stream.Protocol)
80
- }
81
- return res, nil
61
+ return c.streams, nil
62
}
63
64
func (api *SwarmAPI) Peers(ctx context.Context) ([]iface.ConnectionInfo, error) {
85
- var out struct {
86
- Peers []*connInfo
65
+ var resp struct {
66
+ Peers []struct{
67
+ Addr string
68
+ Peer string
69
+ Latency time.Duration
70
+ Muxer string
71
+ Direction inet.Direction
72
+ Streams []struct {
73
+ Protocol string
74
+ }
75
+ }
76
}
77
78
err := api.core().request("swarm/peers").
79
Option("streams", true).
80
Option("latency", true).
92
- Exec(ctx, &out)
81
+ Exec(ctx, &resp)
82
if err != nil {
83
return nil, err
84
}
85
97
- res := make([]iface.ConnectionInfo, len(out.Peers))
98
- for i, conn := range out.Peers {
99
- if err := conn.valid(); err != nil {
86
+ res := make([]iface.ConnectionInfo, len(resp.Peers))
87
+ for i, conn := range resp.Peers {
88
+ out := &connInfo{
89
+ latency: conn.Latency,
90
+ muxer: conn.Muxer,
91
+ direction: conn.Direction,
92
+ }
93
+
94
+ out.peer, err = peer.IDB58Decode(conn.Peer)
95
+ if err != nil {
96
return nil, err
97
}
102
- res[i] = conn
98
+
99
+ out.addr, err = multiaddr.NewMultiaddr(conn.Addr)
100
+ if err != nil {
101
+ return nil, err
102
+ }
103
+
104
+ out.streams = make([]protocol.ID, len(conn.Streams))
105
+ for i, p := range conn.Streams {
106
+ out.streams[i] = protocol.ID(p.Protocol)
107
+ }
108
+
109
+ res[i] = out
110
}
111
112
return res, nil