Skip to content

Commit

Permalink
PTP API: Address review comments
Browse files Browse the repository at this point in the history 
License: MIT
Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
  • Loading branch information
@magik6k
magik6k committed Jun 7, 2017
1 parent 104268f commit 043aa5f
Show file tree
Hide file tree
Showing 5 changed files with 152 additions and 134 deletions.
201 changes: 102 additions & 99 deletions core/commands/ptp.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -51,39 +51,54 @@ Note: this command is experimental and subject to change as usecases and APIs ar
},

Subcommands: map[string]*cmds.Command{
"ls": ptpLsCmd,
"streams": ptpStreamsCmd,
"dial": ptpDialCmd,
"listen": ptpListenCmd,
"close": ptpCloseCmd,
"listener": ptpListenerCmd,
"stream": ptpStreamCmd,
},
}

var ptpLsCmd = &cmds.Command{
// ptpListenerCmd is the 'ipfs ptp listener' command
var ptpListenerCmd = &cmds.Command{
Helptext: cmds.HelpText{
Tagline: "P2P listener management.",
ShortDescription: "Create and manage listener p2p endpoints",
},

Subcommands: map[string]*cmds.Command{
"ls": ptpListenerLsCmd,
"open": ptpListenerListenCmd,
"close": ptpListenerCloseCmd,
},
}

// ptpStreamCmd is the 'ipfs ptp stream' command
var ptpStreamCmd = &cmds.Command{
Helptext: cmds.HelpText{
Tagline: "P2P stream management.",
ShortDescription: "Create and manage p2p streams",
},

Subcommands: map[string]*cmds.Command{
"ls": ptpStreamLsCmd,
"dial": ptpStreamDialCmd,
"close": ptpStreamCloseCmd,
},
}

var ptpListenerLsCmd = &cmds.Command{
Helptext: cmds.HelpText{
Tagline: "List active p2p listeners.",
},
Options: []cmds.Option{
cmds.BoolOption("headers", "v", "Print table headers (HandlerID, Protocol, Local, Remote).").Default(false),
},
Run: func(req cmds.Request, res cmds.Response) {
n, err := req.InvocContext().GetNode()
if err != nil {
res.SetError(err, cmds.ErrNormal)
return
}

err = checkEnabled(n)
n, err := getNode(req)
if err != nil {
res.SetError(err, cmds.ErrNormal)
return
}

if !n.OnlineMode() {
res.SetError(errNotOnline, cmds.ErrClient)
return
}

output := &PTPLsOutput{}

for _, listener := range n.PTP.Listeners.Listeners {
Expand Down Expand Up @@ -116,31 +131,20 @@ var ptpLsCmd = &cmds.Command{
},
}

var ptpStreamsCmd = &cmds.Command{
var ptpStreamLsCmd = &cmds.Command{
Helptext: cmds.HelpText{
Tagline: "List active p2p streams.",
},
Options: []cmds.Option{
cmds.BoolOption("headers", "v", "Print table headers (HagndlerID, Protocol, Local, Remote).").Default(false),
},
Run: func(req cmds.Request, res cmds.Response) {
n, err := req.InvocContext().GetNode()
n, err := getNode(req)
if err != nil {
res.SetError(err, cmds.ErrNormal)
return
}

err = checkEnabled(n)
if err != nil {
res.SetError(err, cmds.ErrNormal)
return
}

if !n.OnlineMode() {
res.SetError(errNotOnline, cmds.ErrClient)
return
}

output := &PTPStreamsOutput{}

for _, s := range n.PTP.Streams.Streams {
Expand Down Expand Up @@ -180,7 +184,7 @@ var ptpStreamsCmd = &cmds.Command{
},
}

var ptpListenCmd = &cmds.Command{
var ptpListenerListenCmd = &cmds.Command{
Helptext: cmds.HelpText{
Tagline: "Forward p2p connections to a network multiaddr.",
ShortDescription: `
Expand All @@ -194,23 +198,12 @@ Note that the connections originate from the ipfs daemon process.
cmds.StringArg("Address", true, false, "Request handling application address."),
},
Run: func(req cmds.Request, res cmds.Response) {
n, err := req.InvocContext().GetNode()
if err != nil {
res.SetError(err, cmds.ErrNormal)
return
}

err = checkEnabled(n)
n, err := getNode(req)
if err != nil {
res.SetError(err, cmds.ErrNormal)
return
}

if !n.OnlineMode() {
res.SetError(errNotOnline, cmds.ErrClient)
return
}

proto := "/ptp/" + req.Arguments()[0]
if n.PTP.CheckProtoExists(proto) {
res.SetError(errors.New("protocol handler already registered"), cmds.ErrNormal)
Expand All @@ -237,7 +230,7 @@ Note that the connections originate from the ipfs daemon process.
},
}

var ptpDialCmd = &cmds.Command{
var ptpStreamDialCmd = &cmds.Command{
Helptext: cmds.HelpText{
Tagline: "Dial to a p2p listener.",

Expand All @@ -255,23 +248,12 @@ transparently connect to a p2p service.
cmds.StringArg("BindAddress", false, false, "Address to listen for connection/s (default: /ip4/127.0.0.1/tcp/0)."),
},
Run: func(req cmds.Request, res cmds.Response) {
n, err := req.InvocContext().GetNode()
n, err := getNode(req)
if err != nil {
res.SetError(err, cmds.ErrNormal)
return
}

err = checkEnabled(n)
if err != nil {
res.SetError(err, cmds.ErrNormal)
return
}

if !n.OnlineMode() {
res.SetError(errNotOnline, cmds.ErrClient)
return
}

addr, peer, err := ParsePeerParam(req.Arguments()[0])
if err != nil {
res.SetError(err, cmds.ErrNormal)
Expand Down Expand Up @@ -304,89 +286,110 @@ transparently connect to a p2p service.
},
}

var ptpCloseCmd = &cmds.Command{
var ptpListenerCloseCmd = &cmds.Command{
Helptext: cmds.HelpText{
Tagline: "Closes an active p2p stream or listener.",
Tagline: "Close active p2p listener.",
},
Arguments: []cmds.Argument{
cmds.StringArg("Identifier", false, false, "Stream HandlerID or p2p listener protocol"),
cmds.StringArg("Protocol", false, false, "P2P listener protocol"),
},
Options: []cmds.Option{
cmds.BoolOption("all", "a", "Close all streams and listeners.").Default(false),
cmds.BoolOption("all", "a", "Close all listeners.").Default(false),
},
Run: func(req cmds.Request, res cmds.Response) {
n, err := req.InvocContext().GetNode()
n, err := getNode(req)
if err != nil {
res.SetError(err, cmds.ErrNormal)
return
}

err = checkEnabled(n)
if err != nil {
res.SetError(err, cmds.ErrNormal)
return
closeAll, _, _ := req.Option("all").Bool()
var proto string

if !closeAll {
if len(req.Arguments()) == 0 {
res.SetError(errors.New("no protocol name specified"), cmds.ErrNormal)
return
}

proto = "/ptp/" + req.Arguments()[0]
}

if !n.OnlineMode() {
res.SetError(errNotOnline, cmds.ErrClient)
for _, listener := range n.PTP.Listeners.Listeners {
if !closeAll && listener.Protocol != proto {
continue
}
listener.Close()
if !closeAll {
break
}
}
},
}

var ptpStreamCloseCmd = &cmds.Command{
Helptext: cmds.HelpText{
Tagline: "Close active p2p stream.",
},
Arguments: []cmds.Argument{
cmds.StringArg("HandlerID", false, false, "Stream HandlerID"),
},
Options: []cmds.Option{
cmds.BoolOption("all", "a", "Close all streams.").Default(false),
},
Run: func(req cmds.Request, res cmds.Response) {
n, err := getNode(req)
if err != nil {
res.SetError(err, cmds.ErrNormal)
return
}

closeAll, _, _ := req.Option("all").Bool()

var proto string
var handlerID uint64

useHandlerID := false

if !closeAll {
if len(req.Arguments()) == 0 {
res.SetError(errors.New("no handlerID nor listener protocol specified"), cmds.ErrNormal)
res.SetError(errors.New("no HandlerID specified"), cmds.ErrNormal)
return
}

handlerID, err = strconv.ParseUint(req.Arguments()[0], 10, 64)
if err != nil {
proto = "/ptp/" + req.Arguments()[0]
} else {
useHandlerID = true
res.SetError(err, cmds.ErrNormal)
return
}
}

if closeAll || useHandlerID {
for _, stream := range n.PTP.Streams.Streams {
if !closeAll && handlerID != stream.HandlerID {
continue
}
stream.Close()
if !closeAll {
break
}
for _, stream := range n.PTP.Streams.Streams {
if !closeAll && handlerID != stream.HandlerID {
continue
}
}

if closeAll || !useHandlerID {
for _, listener := range n.PTP.Listeners.Listeners {
if !closeAll && listener.Protocol != proto {
continue
}
listener.Close()
if !closeAll {
break
}
stream.Close()
if !closeAll {
break
}
}
},
}

func checkEnabled(n *core.IpfsNode) error {
func getNode(req cmds.Request) (*core.IpfsNode, error) {
n, err := req.InvocContext().GetNode()
if err != nil {
return nil, err
}

config, err := n.Repo.Config()
if err != nil {
return err
return nil, err
}

if !config.Experimental.Libp2pStreamMounting {
return errors.New("libp2p stream mounting not enabled")
return nil, errors.New("libp2p stream mounting not enabled")
}
return nil

if !n.OnlineMode() {
return nil, errNotOnline
}

return n, nil
}
1 change: 1 addition & 0 deletions ptp/ptp.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ func (ptp *PTP) newStreamTo(ctx2 context.Context, p peer.ID, protocol string) (n
return ptp.peerHost.NewStream(ctx2, p, pro.ID(protocol))
}

// Dial creates new P2P stream to a remote listener
func (ptp *PTP) Dial(ctx context.Context, addr ma.Multiaddr, peer peer.ID, proto string, bindAddr ma.Multiaddr) (*ListenerInfo, error) {
lnet, _, err := manet.DialArgs(bindAddr)
if err != nil {
Expand Down
8 changes: 4 additions & 4 deletions ptp/registry.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -83,10 +83,10 @@ type StreamInfo struct {
}

// Close closes stream endpoints and deregisters it
func (c *StreamInfo) Close() error {
c.Local.Close()
c.Remote.Close()
c.Registry.Deregister(c.HandlerID)
func (s *StreamInfo) Close() error {
s.Local.Close()
s.Remote.Close()
s.Registry.Deregister(s.HandlerID)
return nil
}

Expand Down
7 changes: 5 additions & 2 deletions test/dependencies/ma-pipe-unidir/main.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,11 @@ func app() int {
mode := args[0]
addr := args[1]

if mode != "send" && mode != "recv" {
fmt.Print(USAGE)
return 1
}

if len(opts.PidFile) > 0 {
data := []byte(strconv.Itoa(os.Getpid()))
err := ioutil.WriteFile(opts.PidFile, data, 0644)
Expand Down Expand Up @@ -80,8 +85,6 @@ func app() int {
case "send":
io.Copy(conn, os.Stdin)
default:
//TODO: a bit late
fmt.Print(USAGE)
return 1
}
return 0
Expand Down
Loading

0 comments on commit 043aa5f

Please sign in to comment.