forked from nanomsg/mangos-v1
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fixes nanomsg#2 Add support for Named Pipes on Windows
This adds support for Windows Named Pipes as ipc:// URLs that are compatible with nanomsg and NNG. It also includes (untested) support for SecurityDescriptor and InputBufferSize and OutputBufferSize tunables. It is built on the github.com/Microsoft/go-winio library. Note that legacy libnanomsg (all versions 1.1.3 and earlier) is very fragile in it's handling of IPC, and assumes that senders will only send a single atomic write. This assumption requires us to make an extra data copy. (Note that NNG has no such assumptions, and we could easily dispense with the data copy for NNG.)
- Loading branch information
Showing
9 changed files
with
419 additions
and
103 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| // +build !windows | ||
|
|
||
| // Copyright 2018 The Mangos Authors | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use file except in compliance with the License. | ||
| // You may obtain a copy of the license at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package mangos | ||
|
|
||
| import ( | ||
| "encoding/binary" | ||
| "io" | ||
| "net" | ||
| ) | ||
|
|
||
| // NewConnPipeIPC allocates a new Pipe using the IPC exchange protocol. | ||
| func NewConnPipeIPC(c net.Conn, sock Socket, props ...interface{}) (Pipe, error) { | ||
| p := &connipc{conn: conn{c: c, proto: sock.GetProtocol(), sock: sock}} | ||
|
|
||
| if err := p.handshake(props); err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return p, nil | ||
| } | ||
|
|
||
| func (p *connipc) Send(msg *Message) error { | ||
|
|
||
| l := uint64(len(msg.Header) + len(msg.Body)) | ||
| // one := [1]byte{1} | ||
| var err error | ||
|
|
||
| // send length header | ||
| header := make([]byte, 9) | ||
| header[0] = 1 | ||
| binary.BigEndian.PutUint64(header[1:], l) | ||
|
|
||
| if _, err = p.c.Write(header[:]); err != nil { | ||
| return err | ||
| } | ||
|
|
||
| if _, err = p.c.Write(msg.Header); err != nil { | ||
| return err | ||
| } | ||
| // hope this works | ||
| if _, err = p.c.Write(msg.Body); err != nil { | ||
| return err | ||
| } | ||
| msg.Free() | ||
| return nil | ||
| } | ||
|
|
||
| func (p *connipc) Recv() (*Message, error) { | ||
|
|
||
| var sz int64 | ||
| var err error | ||
| var msg *Message | ||
| var one [1]byte | ||
|
|
||
| if _, err = p.c.Read(one[:]); err != nil { | ||
| return nil, err | ||
| } | ||
| if err = binary.Read(p.c, binary.BigEndian, &sz); err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| // Limit messages to the maximum receive value, if not | ||
| // unlimited. This avoids a potential denaial of service. | ||
| if sz < 0 || (p.maxrx > 0 && sz > p.maxrx) { | ||
| return nil, ErrTooLong | ||
| } | ||
| msg = NewMessage(int(sz)) | ||
| msg.Body = msg.Body[0:sz] | ||
| if _, err = io.ReadFull(p.c, msg.Body); err != nil { | ||
| msg.Free() | ||
| return nil, err | ||
| } | ||
| return msg, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,87 @@ | ||
| // +build windows | ||
|
|
||
| // Copyright 2018 The Mangos Authors | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use file except in compliance with the License. | ||
| // You may obtain a copy of the license at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package mangos | ||
|
|
||
| import ( | ||
| "encoding/binary" | ||
| "io" | ||
| "net" | ||
| ) | ||
|
|
||
| // NewConnPipeIPC allocates a new Pipe using the IPC exchange protocol. | ||
| func NewConnPipeIPC(c net.Conn, sock Socket, props ...interface{}) (Pipe, error) { | ||
| p := &connipc{conn: conn{c: c, proto: sock.GetProtocol(), sock: sock}} | ||
|
|
||
| if err := p.handshake(props); err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return p, nil | ||
| } | ||
|
|
||
| func (p *connipc) Send(msg *Message) error { | ||
|
|
||
| l := uint64(len(msg.Header) + len(msg.Body)) | ||
| var err error | ||
|
|
||
| // On Windows, we have to put everything into a contiguous buffer. | ||
| // This is to workaround bugs in legacy libnanomsg. Eventually we | ||
| // might do away with this logic, but only when legacy libnanomsg | ||
| // has been fixed. This puts some pressure on the gc too, which | ||
| // makes me pretty sad. | ||
|
|
||
| // send length header | ||
| buf := make([]byte, 9, 9+l) | ||
| buf[0] = 1 | ||
| binary.BigEndian.PutUint64(buf[1:], l) | ||
| buf = append(buf, msg.Header...) | ||
| buf = append(buf, msg.Body...) | ||
|
|
||
| if _, err = p.c.Write(buf[:]); err != nil { | ||
| return err | ||
| } | ||
| msg.Free() | ||
| return nil | ||
| } | ||
|
|
||
| func (p *connipc) Recv() (*Message, error) { | ||
|
|
||
| var sz int64 | ||
| var err error | ||
| var msg *Message | ||
| var one [1]byte | ||
|
|
||
| if _, err = p.c.Read(one[:]); err != nil { | ||
| return nil, err | ||
| } | ||
| if err = binary.Read(p.c, binary.BigEndian, &sz); err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| // Limit messages to the maximum receive value, if not | ||
| // unlimited. This avoids a potential denaial of service. | ||
| if sz < 0 || (p.maxrx > 0 && sz > p.maxrx) { | ||
| return nil, ErrTooLong | ||
| } | ||
| msg = NewMessage(int(sz)) | ||
| msg.Body = msg.Body[0:sz] | ||
| if _, err = io.ReadFull(p.c, msg.Body); err != nil { | ||
| msg.Free() | ||
| return nil, err | ||
| } | ||
| return msg, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.