remove zmq from project
All checks were successful
continuous-integration/drone/push Build is passing

This commit is contained in:
Paul 2023-03-05 22:37:17 +01:00
parent 98f07aaff8
commit bb76849f33
34 changed files with 6 additions and 3980 deletions

1
.gitignore vendored
View File

@ -2,4 +2,5 @@
*.ini
*.swp
/test
/old
version.go

View File

@ -10,7 +10,6 @@ import (
"git.paulbsd.com/paulbsd/ipbl/src/database"
"git.paulbsd.com/paulbsd/ipbl/src/models"
"git.paulbsd.com/paulbsd/ipbl/src/routers"
"git.paulbsd.com/paulbsd/ipbl/src/zmqrouter"
"git.paulbsd.com/paulbsd/ipbl/utils"
_ "github.com/lib/pq"
)
@ -36,7 +35,7 @@ func main() {
if !cfg.Switchs.NoScanIP {
go models.ScanIP(&cfg)
}
go zmqrouter.Init(&cfg)
//go zmqrouter.Init(&cfg)
go func() { err = routers.RunServer(&ctx, &cfg) }()
if err != nil {
log.Fatalln(err)

1
go.mod
View File

@ -7,7 +7,6 @@ require (
github.com/lib/pq v1.10.7
golang.org/x/net v0.8.0
gopkg.in/ini.v1 v1.67.0
gopkg.in/zeromq/goczmq.v4 v4.1.0
xorm.io/xorm v1.3.2
)

2
go.sum
View File

@ -572,8 +572,6 @@ gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/zeromq/goczmq.v4 v4.1.0 h1:CE+FE81mGVs2aSlnbfLuS1oAwdcVywyMM2AC1g33imI=
gopkg.in/zeromq/goczmq.v4 v4.1.0/go.mod h1:h4IlfePEYMpFdywGr5gAwKhBBj+hiBl/nF4VoSE4k+0=
honnef.co/go/tools v0.0.0-20180728063816-88497007e858/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=

View File

@ -48,7 +48,7 @@ func (cfg *Config) GetConfig() error {
cfg.DbParams.DbPassword = ipblsection.Key("password").MustString("password")
cfg.DbParams.DbDatabase = ipblsection.Key("database").MustString("database")
cfg.Options.ZMQChannel = "ipbl"
cfg.Options.SocketChannel = "ipbl"
cfg.Options.HideBanner = ipblsection.Key("hidebanner").MustBool(false)
return nil
@ -63,9 +63,9 @@ type Config struct {
DbDatabase string
}
Options struct {
Version string
HideBanner bool
ZMQChannel string
Version string
HideBanner bool
SocketChannel string
}
Switchs struct {
Port int

View File

@ -23,7 +23,6 @@ func init() {
new(Cfg),
new(CfgSet),
new(CfgTrustlist),
new(CfgZMQ),
new(CfgWS),
new(City),
new(Country),

View File

@ -1,110 +0,0 @@
package zmqrouter
import (
"encoding/json"
"fmt"
"log"
"time"
"git.paulbsd.com/paulbsd/ipbl/src/config"
"git.paulbsd.com/paulbsd/ipbl/src/models"
"gopkg.in/zeromq/goczmq.v4"
)
func Init(cfg *config.Config) (err error) {
for {
log.Println("Initiating ZMQ sockets")
reqsock, err := InitRep()
if err != nil {
break
}
pubsock, err := InitPub()
if err != nil {
break
}
err = Handle(cfg, reqsock, pubsock, cfg.Options.ZMQChannel)
if err != nil {
reqsock.Destroy()
pubsock.Destroy()
time.Sleep(1 * time.Second)
break
}
}
return
}
func Handle(cfg *config.Config, reqsock *goczmq.Sock, pubsock *goczmq.Sock, channel string) (err error) {
log.Println("Start handling zmq sockets")
var lastip string
for {
var msg = "err"
var req, err = reqsock.RecvMessage()
if err != nil {
log.Println("unable to receive message from req socket", err)
continue
}
var topub [][]byte
for _, val := range req {
var apievent = models.APIEvent{}
var event = models.Event{}
err = json.Unmarshal(val, &apievent)
if err != nil {
log.Println("unable to parse ip address", err)
time.Sleep(time.Second)
continue
}
if apievent.IPData.IP != "" && apievent.IPData.IP == lastip {
continue
}
if apievent.MsgType == "add" {
session := cfg.Db.NewSession()
event.APIParse(session, apievent)
session.Close()
err := event.Insert(cfg)
if err != nil {
log.Println(err)
}
log.Printf("zmq: Inserted event")
}
if apievent.MsgType == "file" {
apievent.MsgType = "zmq"
}
val, err = json.Marshal(apievent)
if err != nil {
fmt.Println(err)
}
tmpval := fmt.Sprintf("%s %s", channel, string(val))
ipjson := []byte(tmpval)
topub = append(topub, ipjson)
lastip = apievent.IPData.IP
}
err = pubsock.SendMessage(topub)
if err != nil {
log.Println("error sending message to pub socket")
continue
}
msg = "ok"
var resp [][]byte = [][]byte{[]byte(msg)}
err = reqsock.SendMessage(resp)
if err != nil {
log.Println("error replying message to req socket")
continue
}
}
}

View File

@ -1,14 +0,0 @@
package zmqrouter
import (
"fmt"
"gopkg.in/zeromq/goczmq.v4"
)
const pubsubport int = 9999
func InitPub() (sock *goczmq.Sock, err error) {
sock, err = goczmq.NewPub(fmt.Sprintf("tcp://*:%d", pubsubport))
return
}

View File

@ -1,14 +0,0 @@
package zmqrouter
import (
"fmt"
"gopkg.in/zeromq/goczmq.v4"
)
const repreqport int = 9998
func InitRep() (sock *goczmq.Sock, err error) {
sock, err = goczmq.NewRep(fmt.Sprintf("tcp://*:%d", repreqport))
return
}

View File

@ -1,23 +0,0 @@
# Compiled Object files, Static and Dynamic libs (Shared Objects)
*.o
*.a
*.so
# Folders
_obj
_test
# Architecture specific extensions/prefixes
*.[568vq]
[568vq].out
*.cgo1.go
*.cgo2.c
_cgo_defun.c
_cgo_gotypes.go
_cgo_export.*
_testmain.go
*.exe
*.test

View File

@ -1,17 +0,0 @@
language: go
warnings_are_errors: false
addons:
apt:
sources:
- sourceline: 'deb http://download.opensuse.org/repositories/network:/messaging:/zeromq:/git-draft/xUbuntu_12.04/ ./'
key_url: 'http://download.opensuse.org/repositories/network:/messaging:/zeromq:/git-draft/xUbuntu_12.04/Release.key'
packages:
- libsodium-dev
- libczmq-dev
go:
- 1.7.1
script:
- go get -t -v ./...
- go test -v .

View File

@ -1,15 +0,0 @@
Contributors (no particular order)
==================================
Brian Knox
Luna Duclos
Indradhanush Gupta
Armen Baghumian
Ben Aldrich
Jordan Sissel
Nick Bargnesi
Cyrille Verrier
Ben Aldrich
Matthew Campbell
James Reuss
Pieter Hintjens

View File

@ -1,22 +0,0 @@
# Contributing to GoCZMQ
The contributors are listed in AUTHORS (add yourself). This project uses the MPL v2 license, see LICENSE.
# Our Process
Before you send a pull request, please familiarize yourself with the [C4.1 Collective Code Construction Contract](http://rfc.zeromq.org/spec:22) process. A quick summary (but please, do read the process document):
* A Pull Request should be described in the form of a problem statement.
* The code included with the pull request should be a proposed solution to that problem.
* The submitted code should adhere to our style guidelines (described below).
* The submitted code should include tests.
* The submitted code should not break any existing tests.
"A Problem" should be one single clear problem. Large complex problems should be broken down into a series of smaller problems when ever possible.
**Please be aware** that GoCZMQ is **not versioned**. We merge to master. We deploy from master. Master is epxected to be working, at all times. We strive to do our very best to never break public API in this library. Changes can be additive, but they can not break the existing API. If a case arises where we need to, we will be loud about it on the ZeroMQ mailing list and try to build consensus among current maintainers that it's necessary. We will be very chagrined about it, and you can poke fun at us a bit.
# Style Guide
* Your code must be formatted with [Gofmt](https://blog.golang.org/go-fmt-your-code)
* Your code should pass [golint](https://github.com/golang/lint). If for some reason it cannot, please provide an explanation.
* Your code should pass [go vet](https://golang.org/cmd/vet/)

View File

@ -1,362 +0,0 @@
Mozilla Public License, version 2.0
1. Definitions
1.1. "Contributor"
means each individual or legal entity that creates, contributes to the
creation of, or owns Covered Software.
1.2. "Contributor Version"
means the combination of the Contributions of others (if any) used by a
Contributor and that particular Contributor's Contribution.
1.3. "Contribution"
means Covered Software of a particular Contributor.
1.4. "Covered Software"
means Source Code Form to which the initial Contributor has attached the
notice in Exhibit A, the Executable Form of such Source Code Form, and
Modifications of such Source Code Form, in each case including portions
thereof.
1.5. "Incompatible With Secondary Licenses"
means
a. that the initial Contributor has attached the notice described in
Exhibit B to the Covered Software; or
b. that the Covered Software was made available under the terms of
version 1.1 or earlier of the License, but not also under the terms of
a Secondary License.
1.6. "Executable Form"
means any form of the work other than Source Code Form.
1.7. "Larger Work"
means a work that combines Covered Software with other material, in a
separate file or files, that is not Covered Software.
1.8. "License"
means this document.
1.9. "Licensable"
means having the right to grant, to the maximum extent possible, whether
at the time of the initial grant or subsequently, any and all of the
rights conveyed by this License.
1.10. "Modifications"
means any of the following:
a. any file in Source Code Form that results from an addition to,
deletion from, or modification of the contents of Covered Software; or
b. any new file in Source Code Form that contains any Covered Software.
1.11. "Patent Claims" of a Contributor
means any patent claim(s), including without limitation, method,
process, and apparatus claims, in any patent Licensable by such
Contributor that would be infringed, but for the grant of the License,
by the making, using, selling, offering for sale, having made, import,
or transfer of either its Contributions or its Contributor Version.
1.12. "Secondary License"
means either the GNU General Public License, Version 2.0, the GNU Lesser
General Public License, Version 2.1, the GNU Affero General Public
License, Version 3.0, or any later versions of those licenses.
1.13. "Source Code Form"
means the form of the work preferred for making modifications.
1.14. "You" (or "Your")
means an individual or a legal entity exercising rights under this
License. For legal entities, "You" includes any entity that controls, is
controlled by, or is under common control with You. For purposes of this
definition, "control" means (a) the power, direct or indirect, to cause
the direction or management of such entity, whether by contract or
otherwise, or (b) ownership of more than fifty percent (50%) of the
outstanding shares or beneficial ownership of such entity.
2. License Grants and Conditions
2.1. Grants
Each Contributor hereby grants You a world-wide, royalty-free,
non-exclusive license:
a. under intellectual property rights (other than patent or trademark)
Licensable by such Contributor to use, reproduce, make available,
modify, display, perform, distribute, and otherwise exploit its
Contributions, either on an unmodified basis, with Modifications, or
as part of a Larger Work; and
b. under Patent Claims of such Contributor to make, use, sell, offer for
sale, have made, import, and otherwise transfer either its
Contributions or its Contributor Version.
2.2. Effective Date
The licenses granted in Section 2.1 with respect to any Contribution
become effective for each Contribution on the date the Contributor first
distributes such Contribution.
2.3. Limitations on Grant Scope
The licenses granted in this Section 2 are the only rights granted under
this License. No additional rights or licenses will be implied from the
distribution or licensing of Covered Software under this License.
Notwithstanding Section 2.1(b) above, no patent license is granted by a
Contributor:
a. for any code that a Contributor has removed from Covered Software; or
b. for infringements caused by: (i) Your and any other third party's
modifications of Covered Software, or (ii) the combination of its
Contributions with other software (except as part of its Contributor
Version); or
c. under Patent Claims infringed by Covered Software in the absence of
its Contributions.
This License does not grant any rights in the trademarks, service marks,
or logos of any Contributor (except as may be necessary to comply with
the notice requirements in Section 3.4).
2.4. Subsequent Licenses
No Contributor makes additional grants as a result of Your choice to
distribute the Covered Software under a subsequent version of this
License (see Section 10.2) or under the terms of a Secondary License (if
permitted under the terms of Section 3.3).
2.5. Representation
Each Contributor represents that the Contributor believes its
Contributions are its original creation(s) or it has sufficient rights to
grant the rights to its Contributions conveyed by this License.
2.6. Fair Use
This License is not intended to limit any rights You have under
applicable copyright doctrines of fair use, fair dealing, or other
equivalents.
2.7. Conditions
Sections 3.1, 3.2, 3.3, and 3.4 are conditions of the licenses granted in
Section 2.1.
3. Responsibilities
3.1. Distribution of Source Form
All distribution of Covered Software in Source Code Form, including any
Modifications that You create or to which You contribute, must be under
the terms of this License. You must inform recipients that the Source
Code Form of the Covered Software is governed by the terms of this
License, and how they can obtain a copy of this License. You may not
attempt to alter or restrict the recipients' rights in the Source Code
Form.
3.2. Distribution of Executable Form
If You distribute Covered Software in Executable Form then:
a. such Covered Software must also be made available in Source Code Form,
as described in Section 3.1, and You must inform recipients of the
Executable Form how they can obtain a copy of such Source Code Form by
reasonable means in a timely manner, at a charge no more than the cost
of distribution to the recipient; and
b. You may distribute such Executable Form under the terms of this
License, or sublicense it under different terms, provided that the
license for the Executable Form does not attempt to limit or alter the
recipients' rights in the Source Code Form under this License.
3.3. Distribution of a Larger Work
You may create and distribute a Larger Work under terms of Your choice,
provided that You also comply with the requirements of this License for
the Covered Software. If the Larger Work is a combination of Covered
Software with a work governed by one or more Secondary Licenses, and the
Covered Software is not Incompatible With Secondary Licenses, this
License permits You to additionally distribute such Covered Software
under the terms of such Secondary License(s), so that the recipient of
the Larger Work may, at their option, further distribute the Covered
Software under the terms of either this License or such Secondary
License(s).
3.4. Notices
You may not remove or alter the substance of any license notices
(including copyright notices, patent notices, disclaimers of warranty, or
limitations of liability) contained within the Source Code Form of the
Covered Software, except that You may alter any license notices to the
extent required to remedy known factual inaccuracies.
3.5. Application of Additional Terms
You may choose to offer, and to charge a fee for, warranty, support,
indemnity or liability obligations to one or more recipients of Covered
Software. However, You may do so only on Your own behalf, and not on
behalf of any Contributor. You must make it absolutely clear that any
such warranty, support, indemnity, or liability obligation is offered by
You alone, and You hereby agree to indemnify every Contributor for any
liability incurred by such Contributor as a result of warranty, support,
indemnity or liability terms You offer. You may include additional
disclaimers of warranty and limitations of liability specific to any
jurisdiction.
4. Inability to Comply Due to Statute or Regulation
If it is impossible for You to comply with any of the terms of this License
with respect to some or all of the Covered Software due to statute,
judicial order, or regulation then You must: (a) comply with the terms of
this License to the maximum extent possible; and (b) describe the
limitations and the code they affect. Such description must be placed in a
text file included with all distributions of the Covered Software under
this License. Except to the extent prohibited by statute or regulation,
such description must be sufficiently detailed for a recipient of ordinary
skill to be able to understand it.
5. Termination
5.1. The rights granted under this License will terminate automatically if You
fail to comply with any of its terms. However, if You become compliant,
then the rights granted under this License from a particular Contributor
are reinstated (a) provisionally, unless and until such Contributor
explicitly and finally terminates Your grants, and (b) on an ongoing
basis, if such Contributor fails to notify You of the non-compliance by
some reasonable means prior to 60 days after You have come back into
compliance. Moreover, Your grants from a particular Contributor are
reinstated on an ongoing basis if such Contributor notifies You of the
non-compliance by some reasonable means, this is the first time You have
received notice of non-compliance with this License from such
Contributor, and You become compliant prior to 30 days after Your receipt
of the notice.
5.2. If You initiate litigation against any entity by asserting a patent
infringement claim (excluding declaratory judgment actions,
counter-claims, and cross-claims) alleging that a Contributor Version
directly or indirectly infringes any patent, then the rights granted to
You by any and all Contributors for the Covered Software under Section
2.1 of this License shall terminate.
5.3. In the event of termination under Sections 5.1 or 5.2 above, all end user
license agreements (excluding distributors and resellers) which have been
validly granted by You or Your distributors under this License prior to
termination shall survive termination.
6. Disclaimer of Warranty
Covered Software is provided under this License on an "as is" basis,
without warranty of any kind, either expressed, implied, or statutory,
including, without limitation, warranties that the Covered Software is free
of defects, merchantable, fit for a particular purpose or non-infringing.
The entire risk as to the quality and performance of the Covered Software
is with You. Should any Covered Software prove defective in any respect,
You (not any Contributor) assume the cost of any necessary servicing,
repair, or correction. This disclaimer of warranty constitutes an essential
part of this License. No use of any Covered Software is authorized under
this License except under this disclaimer.
7. Limitation of Liability
Under no circumstances and under no legal theory, whether tort (including
negligence), contract, or otherwise, shall any Contributor, or anyone who
distributes Covered Software as permitted above, be liable to You for any
direct, indirect, special, incidental, or consequential damages of any
character including, without limitation, damages for lost profits, loss of
goodwill, work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses, even if such party shall have been
informed of the possibility of such damages. This limitation of liability
shall not apply to liability for death or personal injury resulting from
such party's negligence to the extent applicable law prohibits such
limitation. Some jurisdictions do not allow the exclusion or limitation of
incidental or consequential damages, so this exclusion and limitation may
not apply to You.
8. Litigation
Any litigation relating to this License may be brought only in the courts
of a jurisdiction where the defendant maintains its principal place of
business and such litigation shall be governed by laws of that
jurisdiction, without reference to its conflict-of-law provisions. Nothing
in this Section shall prevent a party's ability to bring cross-claims or
counter-claims.
9. Miscellaneous
This License represents the complete agreement concerning the subject
matter hereof. If any provision of this License is held to be
unenforceable, such provision shall be reformed only to the extent
necessary to make it enforceable. Any law or regulation which provides that
the language of a contract shall be construed against the drafter shall not
be used to construe this License against a Contributor.
10. Versions of the License
10.1. New Versions
Mozilla Foundation is the license steward. Except as provided in Section
10.3, no one other than the license steward has the right to modify or
publish new versions of this License. Each version will be given a
distinguishing version number.
10.2. Effect of New Versions
You may distribute the Covered Software under the terms of the version
of the License under which You originally received the Covered Software,
or under the terms of any subsequent version published by the license
steward.
10.3. Modified Versions
If you create software not governed by this License, and you want to
create a new license for such software, you may create and use a
modified version of this License if you rename the license and remove
any references to the name of the license steward (except to note that
such modified license differs from this License).
10.4. Distributing Source Code Form that is Incompatible With Secondary
Licenses If You choose to distribute Source Code Form that is
Incompatible With Secondary Licenses under the terms of this version of
the License, the notice described in Exhibit B of this License must be
attached.
Exhibit A - Source Code Form License Notice
This Source Code Form is subject to the
terms of the Mozilla Public License, v.
2.0. If a copy of the MPL was not
distributed with this file, You can
obtain one at
http://mozilla.org/MPL/2.0/.
If it is not possible or desirable to put the notice in a particular file,
then You may include the notice in a location (such as a LICENSE file in a
relevant directory) where a recipient would be likely to look for such a
notice.
You may add additional accurate notices of copyright ownership.
Exhibit B - "Incompatible With Secondary Licenses" Notice
This Source Code Form is "Incompatible
With Secondary Licenses", as defined by
the Mozilla Public License, v. 2.0.

View File

@ -1,18 +0,0 @@
build: format clean test
go build ./...
test: get
go test -v .
bench: get
go test -v -bench . ./...
get:
go get -t -v ./...
format:
find . -name \*.go -type f -exec gofmt -w {} \;
clean:
.PHONY: clean build

View File

@ -1,261 +0,0 @@
# goczmq [![Build Status](https://travis-ci.org/zeromq/goczmq.svg?branch=master)](https://travis-ci.org/zeromq/goczmq) [![Doc Status](https://godoc.org/github.com/zeromq/goczmq?status.png)](https://godoc.org/github.com/zeromq/goczmq)
## Introduction
A golang interface to the [CZMQ v4.2](http://czmq.zeromq.org) API.
## Install
### Dependencies
* [libsodium](https://github.com/jedisct1/libsodium)
* [libzmq](https://github.com/zeromq/libzmq)
* [czmq](https://github.com/zeromq/czmq)
### For CZMQ master
```
go get github.com/zeromq/goczmq
```
#### A Note on Build Tags
The CZMQ library includes experimental classes that are not built by default, but can be built
by passing `--enable-drafts` to configure. Support for these draft classes are being added
to goczmq. To build these features against a CZMQ that has been compiled with `--enable-drafts`,
use `go build -tags draft`.
### For CMZQ = 4.0
```
go get gopkg.in/zeromq/goczmq.v4
```
### For CZMQ Before 4.0
```
go get gopkg.in/zeromq/goczmq.v1
```
## Usage
### Direct CZMQ Sock API
#### Example
```go
package main
import (
"log"
"github.com/zeromq/goczmq"
)
func main() {
// Create a router socket and bind it to port 5555.
router, err := goczmq.NewRouter("tcp://*:5555")
if err != nil {
log.Fatal(err)
}
defer router.Destroy()
log.Println("router created and bound")
// Create a dealer socket and connect it to the router.
dealer, err := goczmq.NewDealer("tcp://127.0.0.1:5555")
if err != nil {
log.Fatal(err)
}
defer dealer.Destroy()
log.Println("dealer created and connected")
// Send a 'Hello' message from the dealer to the router.
// Here we send it as a frame ([]byte), with a FlagNone
// flag to indicate there are no more frames following.
err = dealer.SendFrame([]byte("Hello"), goczmq.FlagNone)
if err != nil {
log.Fatal(err)
}
log.Println("dealer sent 'Hello'")
// Receve the message. Here we call RecvMessage, which
// will return the message as a slice of frames ([][]byte).
// Since this is a router socket that support async
// request / reply, the first frame of the message will
// be the routing frame.
request, err := router.RecvMessage()
if err != nil {
log.Fatal(err)
}
log.Printf("router received '%s' from '%v'", request[1], request[0])
// Send a reply. First we send the routing frame, which
// lets the dealer know which client to send the message.
// The FlagMore flag tells the router there will be more
// frames in this message.
err = router.SendFrame(request[0], goczmq.FlagMore)
if err != nil {
log.Fatal(err)
}
log.Printf("router sent 'World'")
// Next send the reply. The FlagNone flag tells the router
// that this is the last frame of the message.
err = router.SendFrame([]byte("World"), goczmq.FlagNone)
if err != nil {
log.Fatal(err)
}
// Receive the reply.
reply, err := dealer.RecvMessage()
if err != nil {
log.Fatal(err)
}
log.Printf("dealer received '%s'", string(reply[0]))
}
```
#### Output
```
2015/05/26 21:52:52 router created and bound
2015/05/26 21:52:52 dealer created and connected
2015/05/26 21:52:52 dealer sent 'Hello'
2015/05/26 21:52:52 router received 'Hello' from '[0 103 84 189 175]'
2015/05/26 21:52:52 router sent 'World'
2015/05/26 21:52:52 dealer received 'World'
```
### io.ReadWriter support
#### Example
```go
package main
import (
"log"
"github.com/zeromq/goczmq"
)
func main() {
// Create a router socket and bind it to port 5555.
router, err := goczmq.NewRouter("tcp://*:5555")
if err != nil {
log.Fatal(err)
}
defer router.Destroy()
log.Println("router created and bound")
// Create a dealer socket and connect it to the router.
dealer, err := goczmq.NewDealer("tcp://127.0.0.1:5555")
if err != nil {
log.Fatal(err)
}
defer dealer.Destroy()
log.Println("dealer created and connected")
// Send a 'Hello' message from the dealer to the router,
// using the io.Write interface
n, err := dealer.Write([]byte("Hello"))
if err != nil {
log.Fatal(err)
}
log.Printf("dealer sent %d byte message 'Hello'\n", n)
// Make a byte slice and pass it to the router
// Read interface. When using the ReadWriter
// interface with a router socket, the router
// caches the routing frames internally in a
// FIFO and uses them transparently when
// sending replies.
buf := make([]byte, 16386)
n, err = router.Read(buf)
if err != nil {
log.Fatal(err)
}
log.Printf("router received '%s'\n", buf[:n])
// Send a reply.
n, err = router.Write([]byte("World"))
if err != nil {
log.Fatal(err)
}
log.Printf("router sent %d byte message 'World'\n", n)
// Receive the reply, reusing the previous buffer.
n, err = dealer.Read(buf)
if err != nil {
log.Fatal(err)
}
log.Printf("dealer received '%s'", string(buf[:n]))
}
```
#### Output
```
2015/05/26 21:54:10 router created and bound
2015/05/26 21:54:10 dealer created and connected
2015/05/26 21:54:10 dealer sent 5 byte message 'Hello'
2015/05/26 21:54:10 router received 'Hello'
2015/05/26 21:54:10 router sent 5 byte message 'World'
2015/05/26 21:54:10 dealer received 'World'
```
### Thread safe channel interface
#### Example
```go
package main
import (
"log"
"github.com/zeromq/goczmq"
)
func main() {
// Create a router channeler and bind it to port 5555.
// A channeler provides a thread safe channel interface
// to a *Sock
router := goczmq.NewRouterChanneler("tcp://*:5555")
defer router.Destroy()
log.Println("router created and bound")
// Create a dealer channeler and connect it to the router.
dealer := goczmq.NewDealerChanneler("tcp://127.0.0.1:5555")
defer dealer.Destroy()
log.Println("dealer created and connected")
// Send a 'Hello' message from the dealer to the router.
dealer.SendChan <- [][]byte{[]byte("Hello")}
log.Println("dealer sent 'Hello'")
// Receve the message as a [][]byte. Since this is
// a router, the first frame of the message wil
// be the routing frame.
request := <-router.RecvChan
log.Printf("router received '%s' from '%v'", request[1], request[0])
// Send a reply. First we send the routing frame, which
// lets the dealer know which client to send the message.
router.SendChan <- [][]byte{request[0], []byte("World")}
log.Printf("router sent 'World'")
// Receive the reply.
reply := <-dealer.RecvChan
log.Printf("dealer received '%s'", string(reply[0]))
}
```
#### Output
```
2015/05/26 21:56:43 router created and bound
2015/05/26 21:56:43 dealer created and connected
2015/05/26 21:56:43 dealer sent 'Hello'
2015/05/26 21:56:43 received 'Hello' from '[0 12 109 153 35]'
2015/05/26 21:56:43 router sent 'World'
2015/05/26 21:56:43 dealer received 'World'
```
## GoDoc
[godoc](https://godoc.org/github.com/zeromq/goczmq)
## See Also
* [Peter Kleiweg's](https://github.com/pebbe) [zmq4](https://github.com/pebbe/zmq4) bindings
## License
This project uses the MPL v2 license, see LICENSE

View File

@ -1,140 +0,0 @@
package goczmq
/*
#include "czmq.h"
zactor_t *Auth_new () {
zactor_t *auth = zactor_new(zauth, NULL); return auth;
}
*/
import "C"
import (
"unsafe"
)
// Auth wraps the CZMQ zauth actor. It handles authentication
// for all incoming connections. It allows whitelisting and
// blackisting peers based on IP address and support
// PLAIN and CURVE authentication policies.
type Auth struct {
zactorT *C.struct__zactor_t
}
// NewAuth creates a new Auth actor.
func NewAuth() *Auth {
z := &Auth{}
z.zactorT = C.Auth_new()
return z
}
// Verbose sets the auth actor to log information to stdout.
func (a *Auth) Verbose() error {
cmd := C.CString("VERBOSE")
defer C.free(unsafe.Pointer(cmd))
rc := C.zstr_send(unsafe.Pointer(a.zactorT), cmd)
if rc == -1 {
return ErrActorCmd
}
C.zsock_wait(unsafe.Pointer(a.zactorT))
return nil
}
// Deny adds an address to a socket's deny list
func (a *Auth) Deny(address string) error {
cmd := C.CString("DENY")
defer C.free(unsafe.Pointer(cmd))
cAddress := C.CString(address)
defer C.free(unsafe.Pointer(cAddress))
rc := C.zstr_sendm(unsafe.Pointer(a.zactorT), cmd)
if rc == -1 {
return ErrActorCmd
}
rc = C.zstr_send(unsafe.Pointer(a.zactorT), cAddress)
if rc == -1 {
return ErrActorCmd
}
C.zsock_wait(unsafe.Pointer(a.zactorT))
return nil
}
// Allow removes a previous Deny
func (a *Auth) Allow(address string) error {
cmd := C.CString("ALLOW")
defer C.free(unsafe.Pointer(cmd))
cAddress := C.CString(address)
defer C.free(unsafe.Pointer(cAddress))
rc := C.zstr_sendm(unsafe.Pointer(a.zactorT), cmd)
if rc == -1 {
return ErrActorCmd
}
rc = C.zstr_send(unsafe.Pointer(a.zactorT), cAddress)
if rc == -1 {
return ErrActorCmd
}
C.zsock_wait(unsafe.Pointer(a.zactorT))
return nil
}
// Curve sets auth method to curve
func (a *Auth) Curve(allowed string) error {
cmd := C.CString("CURVE")
defer C.free(unsafe.Pointer(cmd))
cAllowed := C.CString(allowed)
defer C.free(unsafe.Pointer(cAllowed))
rc := C.zstr_sendm(unsafe.Pointer(a.zactorT), cmd)
if rc == -1 {
return ErrActorCmd
}
rc = C.zstr_send(unsafe.Pointer(a.zactorT), cAllowed)
if rc == -1 {
return ErrActorCmd
}
C.zsock_wait(unsafe.Pointer(a.zactorT))
return nil
}
// Plain sets auth method to plain
func (a *Auth) Plain(directory string) error {
cmd := C.CString("PLAIN")
defer C.free(unsafe.Pointer(cmd))
cDirectory := C.CString(directory)
defer C.free(unsafe.Pointer(cDirectory))
rc := C.zstr_sendm(unsafe.Pointer(a.zactorT), cmd)
if rc == -1 {
return ErrActorCmd
}
rc = C.zstr_send(unsafe.Pointer(a.zactorT), cDirectory)
if rc == -1 {
return ErrActorCmd
}
C.zsock_wait(unsafe.Pointer(a.zactorT))
return nil
}
// Destroy destroys the auth actor.
func (a *Auth) Destroy() {
C.zactor_destroy(&a.zactorT)
}

View File

@ -1,162 +0,0 @@
package goczmq
/*
#include "czmq.h"
zactor_t *Beacon_new () {
zactor_t *beacon = zactor_new(zbeacon, NULL); return beacon;
}
int Beacon_publish(void *actor, void *data, int size, int interval) {
return zsock_send(actor, "sbi", "PUBLISH", (byte*)data, size, interval);
}
*/
import "C"
import (
"strconv"
"unsafe"
)
// Beacon wraps the CZMQ beacon actor. It implements a
// peer-to-peer discovery service for local networks. Beacons
// can broadcast and receive UDPv4 service broadcasts.
type Beacon struct {
zactorT *C.struct__zactor_t
}
// NewBeacon creates a new Beacon instance.
func NewBeacon() *Beacon {
z := &Beacon{}
z.zactorT = C.Beacon_new()
return z
}
// Verbose sets the beacon to log information to stdout.
func (b *Beacon) Verbose() error {
cmd := C.CString("VERBOSE")
defer C.free(unsafe.Pointer(cmd))
rc := C.zstr_send(unsafe.Pointer(b.zactorT), cmd)
if rc == -1 {
return ErrActorCmd
}
return nil
}
// Configure accepts a port number and configures
// the beacon, returning an address
func (b *Beacon) Configure(port int) (string, error) {
cmd := C.CString("CONFIGURE")
defer C.free(unsafe.Pointer(cmd))
cPort := C.CString(strconv.Itoa(port))
defer C.free(unsafe.Pointer(cPort))
rc := C.zstr_sendm(unsafe.Pointer(b.zactorT), cmd)
if rc == -1 {
return "", ErrActorCmd
}
rc = C.zstr_send(unsafe.Pointer(b.zactorT), cPort)
if rc == -1 {
return "", ErrActorCmd
}
cHostname := C.zstr_recv(unsafe.Pointer(b.zactorT))
hostname := C.GoString(cHostname)
return hostname, nil
}
// Publish publishes an announcement string at an interval
func (b *Beacon) Publish(announcement string, interval int) error {
cmd := C.CString("PUBLISH")
defer C.free(unsafe.Pointer(cmd))
cAnnouncement := C.CString(announcement)
defer C.free(unsafe.Pointer(cAnnouncement))
cInterval := C.CString(strconv.Itoa(interval))
defer C.free(unsafe.Pointer(cInterval))
rc := C.zstr_sendm(unsafe.Pointer(b.zactorT), cmd)
if rc == -1 {
return ErrActorCmd
}
rc = C.zstr_sendm(unsafe.Pointer(b.zactorT), cAnnouncement)
if rc == -1 {
return ErrActorCmd
}
rc = C.zstr_send(unsafe.Pointer(b.zactorT), cInterval)
if rc == -1 {
return ErrActorCmd
}
return nil
}
// PublishBytes publishes an announcement byte slice at an interval
func (b *Beacon) PublishBytes(announcement []byte, interval int) error {
rc := C.Beacon_publish(
unsafe.Pointer(b.zactorT),
unsafe.Pointer(&announcement[0]),
C.int(len(announcement)),
C.int(interval),
)
if rc == -1 {
return ErrActorCmd
}
return nil
}
// Subscribe subscribes to beacons matching the filter
func (b *Beacon) Subscribe(filter string) error {
cmd := C.CString("SUBSCRIBE")
defer C.free(unsafe.Pointer(cmd))
cFilter := C.CString(filter)
defer C.free(unsafe.Pointer(cFilter))
rc := C.zstr_sendm(unsafe.Pointer(b.zactorT), cmd)
if rc == -1 {
return ErrActorCmd
}
rc = C.zstr_send(unsafe.Pointer(b.zactorT), cFilter)
if rc == -1 {
return ErrActorCmd
}
return nil
}
// Recv waits for the specific timeout in milliseconds to receive a beacon
func (b *Beacon) Recv(timeout int) [][]byte {
C.zsock_set_rcvtimeo(unsafe.Pointer(b.zactorT), C.int(timeout))
cAddrFrame := C.zframe_recv(unsafe.Pointer(b.zactorT))
defer C.zframe_destroy(&cAddrFrame)
if cAddrFrame == nil {
return nil
}
addr := C.GoBytes(unsafe.Pointer(C.zframe_data(cAddrFrame)), C.int(C.zframe_size(cAddrFrame)))
cBeaconFrame := C.zframe_recv(unsafe.Pointer(b.zactorT))
defer C.zframe_destroy(&cBeaconFrame)
if cBeaconFrame == nil {
return nil
}
beacon := C.GoBytes(unsafe.Pointer(C.zframe_data(cBeaconFrame)), C.int(C.zframe_size(cBeaconFrame)))
return [][]byte{addr, beacon}
}
// Destroy destroys the beacon.
func (b *Beacon) Destroy() {
C.zactor_destroy(&b.zactorT)
}

View File

@ -1,158 +0,0 @@
package goczmq
/*
#include "czmq.h"
void Set_meta(zcert_t *self, const char *key, const char *value) {zcert_set_meta(self, key, "%s", value);}
*/
import "C"
import (
"fmt"
"os"
"unsafe"
)
// Cert wraps the CZMQ zcert class. It provides tools for
// creating and working with ZMQ CURVE security certs.
// The certs can be used as a temporary object in memory
// or persisted to disk. Certs are made up of a public
// and secret keypair + metadata.
type Cert struct {
zcertT *C.struct__zcert_t
}
// NewCert creates a new empty Cert instance
func NewCert() *Cert {
return &Cert{
zcertT: C.zcert_new(),
}
}
// NewCertFromKeys creates a new Cert from a public and private key
func NewCertFromKeys(public []byte, secret []byte) (*Cert, error) {
if len(public) != 32 {
return nil, fmt.Errorf("invalid public key")
}
if len(secret) != 32 {
return nil, fmt.Errorf("invalid private key")
}
return &Cert{
zcertT: C.zcert_new_from(
(*C.byte)(unsafe.Pointer(&public[0])),
(*C.byte)(unsafe.Pointer(&secret[0]))),
}, nil
}
// NewCertFromFile Load loads a Cert from files
func NewCertFromFile(filename string) (*Cert, error) {
_, err := os.Stat(filename)
if os.IsNotExist(err) {
return nil, ErrCertNotFound
}
cFilename := C.CString(filename)
defer C.free(unsafe.Pointer(cFilename))
cert := C.zcert_load(cFilename)
return &Cert{
zcertT: cert,
}, nil
}
// SetMeta sets meta data for a Cert
func (c *Cert) SetMeta(key string, value string) {
cKey := C.CString(key)
defer C.free(unsafe.Pointer(cKey))
cValue := C.CString(value)
defer C.free(unsafe.Pointer(cValue))
C.Set_meta(c.zcertT, cKey, cValue)
}
// Meta returns a meta data item from a Cert given a key
func (c *Cert) Meta(key string) string {
cKey := C.CString(key)
defer C.free(unsafe.Pointer(cKey))
val := C.zcert_meta(c.zcertT, cKey)
return C.GoString(val)
}
// PublicText returns the public key as a string
func (c *Cert) PublicText() string {
val := C.zcert_public_txt(c.zcertT)
return C.GoString(val)
}
// Apply sets the public and private keys for a socket
func (c *Cert) Apply(s *Sock) {
handle := C.zsock_resolve(unsafe.Pointer(s.zsockT))
C.zsock_set_curve_secretkey_bin(handle, C.zcert_secret_key(c.zcertT))
C.zsock_set_curve_publickey_bin(handle, C.zcert_public_key(c.zcertT))
}
// Dup duplicates a Cert
func (c *Cert) Dup() *Cert {
return &Cert{
zcertT: C.zcert_dup(c.zcertT),
}
}
// Equal checks two Certs for equality
func (c *Cert) Equal(compare *Cert) bool {
check := C.zcert_eq(c.zcertT, compare.zcertT)
if check == C.bool(true) {
return true
}
return false
}
// Print prints a Cert to stdout
func (c *Cert) Print() {
C.zcert_print(c.zcertT)
}
// SavePublic saves the public key to a file
func (c *Cert) SavePublic(filename string) error {
cFilename := C.CString(filename)
defer C.free(unsafe.Pointer(cFilename))
rc := C.zcert_save_public(c.zcertT, cFilename)
if rc == C.int(-1) {
return fmt.Errorf("SavePublic error")
}
return nil
}
// SaveSecret saves the secret key to a file
func (c *Cert) SaveSecret(filename string) error {
cFilename := C.CString(filename)
defer C.free(unsafe.Pointer(cFilename))
rc := C.zcert_save_secret(c.zcertT, cFilename)
if rc == C.int(-1) {
return fmt.Errorf("SaveSecret error")
}
return nil
}
// Save saves the public and secret key to filename and filename_secret
func (c *Cert) Save(filename string) error {
cFilename := C.CString(filename)
defer C.free(unsafe.Pointer(cFilename))
rc := C.zcert_save(c.zcertT, cFilename)
if rc == C.int(-1) {
return fmt.Errorf("SavePublic: error")
}
return nil
}
// Destroy destroys Cert instance
func (c *Cert) Destroy() {
C.zcert_destroy(&c.zcertT)
}

View File

@ -1,60 +0,0 @@
package goczmq
/*
#include "czmq.h"
*/
import "C"
import
// CertStore works with directories of CURVE security certificates.
// It lets you easily load stores from disk and check if a key
// is present or not. This could be done fairly easily in pure
// Go, but is included for the sake of compatibility.
"unsafe"
type CertStore struct {
zcertstoreT *C.struct__zcertstore_t
}
// NewCertStore creates a new certificate store from
// a disk directory, loading and indexing all certificates.
func NewCertStore(location string) *CertStore {
cLocation := C.CString(location)
defer C.free(unsafe.Pointer(cLocation))
return &CertStore{
zcertstoreT: C.zcertstore_new(cLocation),
}
}
// Insert inserts a certificate into the store in memory.
// Call Save directly on the cert if you wish to save it
// to disk.
func (c *CertStore) Insert(cert *Cert) {
C.zcertstore_insert(c.zcertstoreT, &cert.zcertT)
}
// Lookup looks up a certificate in the store by public key and
// returns it.
func (c *CertStore) Lookup(key string) *Cert {
cKey := C.CString(key)
defer C.free(unsafe.Pointer(cKey))
ptr := C.zcertstore_lookup(c.zcertstoreT, cKey)
if ptr == nil {
return nil
}
return &Cert{
zcertT: ptr,
}
}
// Print prints a list of certificates in the store to stdout
func (c *CertStore) Print() {
C.zcertstore_print(c.zcertstoreT)
}
// Destroy destroys Cert instance
func (c *CertStore) Destroy() {
C.zcertstore_destroy(&c.zcertstoreT)
}

View File

@ -1,309 +0,0 @@
package goczmq
/*
#include "czmq.h"
void Sock_init() {zsys_init();}
*/
import "C"
import (
"fmt"
"math/rand"
"strings"
)
// Channeler serializes all access to a socket through a send
// and receive channel. It starts two threads, on is used for receiving
// from the zeromq socket. The other is used to listen to the receive
// channel, and send everything back to the socket thrad for sending
// using an additional inproc socket.
type Channeler struct {
id int64
sockType int
endpoints string
subscribe *string
commandAddr string
proxyAddr string
commandChan chan<- string
SendChan chan<- [][]byte
RecvChan <-chan [][]byte
}
// Destroy sends a message to the Channeler to shut it down
// and clean it up.
func (c *Channeler) Destroy() {
c.commandChan <- "destroy"
}
// Subscribe to a Topic
func (c *Channeler) Subscribe(topic string) {
c.commandChan <- fmt.Sprintf("subscribe %s", topic)
}
// Unsubscribe from a Topic
func (c *Channeler) Unsubscribe(topic string) {
c.commandChan <- fmt.Sprintf("unsubscribe %s", topic)
}
// actor is a routine that handles communication with
// the zeromq socket.
func (c *Channeler) actor(recvChan chan<- [][]byte) {
pipe, err := NewPair(fmt.Sprintf(">%s", c.commandAddr))
if err != nil {
panic(err)
}
defer pipe.Destroy()
defer close(recvChan)
pull, err := NewPull(c.proxyAddr)
if err != nil {
panic(err)
}
defer pull.Destroy()
sock := NewSock(c.sockType)
defer sock.Destroy()
switch c.sockType {
case Pub, Rep, Pull, Router, XPub:
err = sock.Attach(c.endpoints, true)
if err != nil {
panic(err)
}
case Req, Push, Dealer, Pair, Stream, XSub:
err = sock.Attach(c.endpoints, false)
if err != nil {
panic(err)
}
case Sub:
if c.subscribe != nil {
subscriptions := strings.Split(*c.subscribe, ",")
for _, topic := range subscriptions {
sock.SetSubscribe(topic)
}
}
err = sock.Attach(c.endpoints, false)
if err != nil {
panic(err)
}
default:
panic(ErrInvalidSockType)
}
poller, err := NewPoller(sock, pull, pipe)
if err != nil {
panic(err)
}
defer poller.Destroy()
for {
s := poller.Wait(-1)
switch s {
case pipe:
cmd, err := pipe.RecvMessage()
if err != nil {
panic(err)
}
switch string(cmd[0]) {
case "destroy":
disconnect := strings.Split(c.endpoints, ",")
for _, endpoint := range disconnect {
sock.Disconnect(endpoint)
}
pipe.SendMessage([][]byte{[]byte("ok")})
goto ExitActor
case "subscribe":
topic := string(cmd[1])
sock.SetSubscribe(topic)
pipe.SendMessage([][]byte{[]byte("ok")})
case "unsubscribe":
topic := string(cmd[1])
sock.SetUnsubscribe(topic)
pipe.SendMessage([][]byte{[]byte("ok")})
}
case sock:
msg, err := s.RecvMessage()
if err != nil {
panic(err)
}
recvChan <- msg
case pull:
msg, err := pull.RecvMessage()
if err != nil {
panic(err)
}
err = sock.SendMessage(msg)
if err != nil {
panic(err)
}
}
}
ExitActor:
}
// channeler is a routine that handles the channel select loop
// and sends commands to the zeromq socket.
func (c *Channeler) channeler(commandChan <-chan string, sendChan <-chan [][]byte) {
push, err := NewPush(c.proxyAddr)
if err != nil {
panic(err)
}
defer push.Destroy()
pipe, err := NewPair(fmt.Sprintf("@%s", c.commandAddr))
if err != nil {
panic(err)
}
defer pipe.Destroy()
for {
select {
case cmd := <-commandChan:
switch cmd {
case "destroy":
err = pipe.SendFrame([]byte("destroy"), FlagNone)
if err != nil {
panic(err)
}
_, err = pipe.RecvMessage()
if err != nil {
panic(err)
}
goto ExitChanneler
default:
parts := strings.Split(cmd, " ")
numParts := len(parts)
message := make([][]byte, numParts, numParts)
for i, p := range parts {
message[i] = []byte(p)
}
err := pipe.SendMessage(message)
if err != nil {
panic(err)
}
_, err = pipe.RecvMessage()
if err != nil {
panic(err)
}
}
case msg := <-sendChan:
err := push.SendMessage(msg)
if err != nil {
panic(err)
}
}
}
ExitChanneler:
}
// newChanneler accepts arguments from the socket type based
// constructors and creates a new Channeler instance
func newChanneler(sockType int, endpoints string, subscribe ...string) *Channeler {
commandChan := make(chan string)
sendChan := make(chan [][]byte)
recvChan := make(chan [][]byte)
C.Sock_init()
c := &Channeler{
id: rand.Int63(),
endpoints: endpoints,
sockType: sockType,
commandChan: commandChan,
SendChan: sendChan,
RecvChan: recvChan,
}
c.commandAddr = fmt.Sprintf("inproc://actorcontrol%d", c.id)
c.proxyAddr = fmt.Sprintf("inproc://proxy%d", c.id)
if len(subscribe) > 0 {
topics := strings.Join(subscribe, ",")
c.subscribe = &topics
}
go c.channeler(commandChan, sendChan)
go c.actor(recvChan)
return c
}
// NewPubChanneler creats a new Channeler wrapping
// a Pub socket. The socket will bind by default.
func NewPubChanneler(endpoints string) *Channeler {
return newChanneler(Pub, endpoints, "")
}
// NewSubChanneler creates a new Channeler wrapping
// a Sub socket. Along with an endpoint list
// it accepts a comma delimited list of topics.
// The socket will connect by default.
func NewSubChanneler(endpoints string, subscribe ...string) *Channeler {
return newChanneler(Sub, endpoints, subscribe...)
}
// NewRepChanneler creates a new Channeler wrapping
// a Rep socket. The socket will bind by default.
func NewRepChanneler(endpoints string) *Channeler {
return newChanneler(Rep, endpoints, "")
}
// NewReqChanneler creates a new Channeler wrapping
// a Req socket. The socket will connect by default.
func NewReqChanneler(endpoints string) *Channeler {
return newChanneler(Req, endpoints, "")
}
// NewPullChanneler creates a new Channeler wrapping
// a Pull socket. The socket will bind by default.
func NewPullChanneler(endpoints string) *Channeler {
return newChanneler(Pull, endpoints, "")
}
// NewPushChanneler creates a new Channeler wrapping
// a Push socket. The socket will connect by default.
func NewPushChanneler(endpoints string) *Channeler {
return newChanneler(Push, endpoints, "")
}
// NewRouterChanneler creates a new Channeler wrapping
// a Router socket. The socket will Bind by default.
func NewRouterChanneler(endpoints string) *Channeler {
return newChanneler(Router, endpoints, "")
}
// NewDealerChanneler creates a new Channeler wrapping
// a Dealer socket. The socket will connect by default.
func NewDealerChanneler(endpoints string) *Channeler {
return newChanneler(Dealer, endpoints, "")
}
// NewXPubChanneler creates a new Channeler wrapping
// an XPub socket. The socket will Bind by default.
func NewXPubChanneler(endpoints string) *Channeler {
return newChanneler(XPub, endpoints, "")
}
// NewXSubChanneler creates a new Channeler wrapping
// a XSub socket. The socket will connect by default.
func NewXSubChanneler(endpoints string) *Channeler {
return newChanneler(XSub, endpoints, "")
}
// NewPairChanneler creates a new Channeler wrapping
// a Pair socket. The socket will connect by default.
func NewPairChanneler(endpoints string) *Channeler {
return newChanneler(Pair, endpoints, "")
}
// NewStreamChanneler creates a new Channeler wrapping
// a Pair socket. The socket will connect by default.
func NewStreamChanneler(endpoints string) *Channeler {
return newChanneler(Stream, endpoints, "")
}

View File

@ -1,201 +0,0 @@
// Package goczmq is a golang binding for CZMQ 3. CZMQ is a high level binding
// for ZeroMQ. Along with ZeroMQ socket support, CZMQ provides "actor" based
// services for authentication, service discovery, and creating proxies.
// GoCZMQ provides direct bindings to CZMQ along with higher level go
// abstractions such as channels and io.ReadWriter interface support.
//
// "Tell them I was a writer.
// A maker of software.
// A humanist. A father.
// And many things.
// But above all, a writer.
// Thank You. :)
// - Pieter Hintjens
package goczmq
/*
#cgo !windows pkg-config: libczmq libzmq libsodium
#cgo windows LDFLAGS: -lws2_32 -liphlpapi -lrpcrt4 -lsodium -lzmq -lczmq
#cgo windows CFLAGS: -Wno-pedantic-ms-format -DLIBCZMQ_EXPORTS -DZMQ_DEFINED_STDINT -DLIBCZMQ_EXPORTS -DZMQ_BUILD_DRAFT_API
#include "czmq.h"
#include <stdlib.h>
#include <string.h>
*/
import "C"
import (
"errors"
)
const (
// Req is a ZMQ_REQ socket type
Req = int(C.ZMQ_REQ)
// Rep is a ZMQ_REP socket type
Rep = int(C.ZMQ_REP)
// Dealer is a ZMQ_DEALER socket type
Dealer = int(C.ZMQ_DEALER)
// Router is a ZMQ_ROUTER socket type
Router = int(C.ZMQ_ROUTER)
// Pub is a ZMQ_PUB socket type
Pub = int(C.ZMQ_PUB)
// Sub is a ZMQ_SUB socket type
Sub = int(C.ZMQ_SUB)
// XPub is a ZMQ_XPUB socket type
XPub = int(C.ZMQ_XPUB)
// XSub is a ZMQ_XSUB socket type
XSub = int(C.ZMQ_XSUB)
// Push is a ZMQ_PUSH socket type
Push = int(C.ZMQ_PUSH)
// Pull is a ZMQ_PULL socket type
Pull = int(C.ZMQ_PULL)
// Pair is a ZMQ_PAIR socket type
Pair = int(C.ZMQ_PAIR)
// Stream is a ZMQ_STREAM socket type
Stream = int(C.ZMQ_STREAM)
// Pollin is the ZMQ_POLLIN constant
Pollin = int(C.ZMQ_POLLIN)
// Pollout is the ZMQ_POLLOUT constant
Pollout = int(C.ZMQ_POLLOUT)
// FlagMore is the ZFRAME_MORE flag
FlagMore = int(C.ZFRAME_MORE)
// FlagReuse is the ZFRAME_REUSE flag
FlagReuse = int(C.ZFRAME_REUSE)
//FlagDontWait is the ZFRAME_DONTWAIT flag
FlagDontWait = int(C.ZFRAME_DONTWAIT)
//FlagNone means there are no flags
FlagNone = 0
// CurveAllowAny is a semantic convenience for allowing
// any Curve clients
CurveAllowAny = "*"
//ZMQVersionMajor is the major version of the underlying ZeroMQ library
ZMQVersionMajor = int(C.ZMQ_VERSION_MAJOR)
//ZMQVersionMinor is the minor version of the underlying ZeroMQ library
ZMQVersionMinor = int(C.ZMQ_VERSION_MINOR)
//CZMQVersionMajor is the major version of the underlying CZMQ library
CZMQVersionMajor = int(C.CZMQ_VERSION_MAJOR)
// CZMQVersionMinor is the minor version of the underlying CZMQ library
CZMQVersionMinor = int(C.CZMQ_VERSION_MINOR)
)
var (
// ErrActorCmd is returned when there is an error sending
// a command to an actor
ErrActorCmd = errors.New("error sending actor command")
// ErrSockAttach is returned when an attach call to a socket fails
ErrSockAttach = errors.New("error attaching zsock")
// ErrInvalidSockType is returned when a function is called
// against a socket type that is not applicable for that socket type
ErrInvalidSockType = errors.New("invalid socket type")
// ErrSliceFull is returned if a []byte passed to Read was not
// large enough to hold the contents of a message
ErrSliceFull = errors.New("slice full")
// ErrConnect is returned if Connect on a socket fails
ErrConnect = errors.New("connect error")
// ErrDisconnect is returned if Disconnect on a socket fails
ErrDisconnect = errors.New("disconnect error")
// ErrBind is returned if Bind on a socket fails
ErrBind = errors.New("bind error")
// ErrUnbind is returned if Unbind on a socket fails
ErrUnbind = errors.New("unbind error")
// ErrSendFrame is returned if SendFrame on a socket fails
ErrSendFrame = errors.New("send frame error")
// ErrRecvFrame is returned if RecvFrame on a socket fails
ErrRecvFrame = errors.New("recv frame error")
// ErrRecvFrameAfterDestroy is returned if RecvFrame is called
// on a socket after it has been destroyed.
ErrRecvFrameAfterDestroy = errors.New("RecvFrame() is invalid on socket after Detroy() has been called.")
// ErrRecvMessage is returned if RecvMessage on a socket fails
ErrRecvMessage = errors.New("recv message error")
// ErrWaitAfterDestroy is returned by a Poller if there is an error
// accessing the underlying socket pointer when Wait is called
ErrWaitAfterDestroy = errors.New("Wait() is invalid on Poller after Destroy() is called.")
// ErrMultiPartUnsupported is returned when a function that does
// not support multi-part messages encounters a multi-part message
ErrMultiPartUnsupported = errors.New("function does not support multi part messages")
// ErrTimeout is returned when a function that supports timeouts times out
ErrTimeout = errors.New("function timed out")
// ErrCertNotFound is returned when NewCertFromFile tries to
// load a file that does not exist.
ErrCertNotFound = errors.New("file not found")
)
// Shutdown shuts down the CZMQ zsys layer.
// The CZMQ zsys layer normally shuts down on process termination through the
// use of an atexit cleanup function. Calling this allows the zsys layer to be
// shutdown manually.
//
// This is beneficial when CZMQ will no longer be used but the process will not
// be terminating. Any potential resources allocated by the zsys layer can be
// freed as they will no longer be needed.
func Shutdown() {
C.zsys_shutdown()
}
func getStringType(k int) string {
switch k {
case Req:
return "REQ"
case Rep:
return "REP"
case Dealer:
return "DEALER"
case Router:
return "ROUTER"
case Pub:
return "PUB"
case Sub:
return "SUB"
case XPub:
return "XPUB"
case XSub:
return "XSUB"
case Push:
return "PUSH"
case Pull:
return "PULL"
case Pair:
return "PAIR"
case Stream:
return "STREAM"
default:
return ""
}
}

View File

@ -1,22 +0,0 @@
<license>
Copyright (C) 2014 the Authors
Permission is hereby granted, free of charge, to any person obtaining
a copy of this software and associated documentation files (the
"Software"), to deal in the Software without restriction, including
without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense, and/or sell copies of the Software, and to
permit persons to whom the Software is furnished to do so, subject to
the following conditions:
The above copyright notice and this permission notice shall be included
in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
</license>

View File

@ -1,106 +0,0 @@
package goczmq
/*
#include "czmq.h"
zpoller_t *Poller_new(void *reader) {
zpoller_t *poller = zpoller_new(reader, NULL);
return poller;
}
*/
import "C"
import (
"fmt"
"unsafe"
)
// Poller provides a simple wrapper to ZeroMQ's zmq_poll API,
// for the common case of reading from a number of sockets.
// Sockets can be added and removed from the running poller.
type Poller struct {
zpollerT *C.struct__zpoller_t
socks []*Sock
}
// NewPoller creates a new Poller instance.
// It accepts one or more readers to poll.
func NewPoller(readers ...*Sock) (*Poller, error) {
var p *Poller
if len(readers) == 0 {
p = &Poller{
zpollerT: C.Poller_new(nil),
socks: make([]*Sock, 0),
}
} else {
p = &Poller{
zpollerT: C.Poller_new(unsafe.Pointer(readers[0].zsockT)),
socks: make([]*Sock, 0),
}
p.socks = append(p.socks, readers[0])
if len(readers) == 1 {
return p, nil
}
for _, reader := range readers[1:] {
err := p.Add(reader)
if err != nil {
return nil, err
}
}
}
return p, nil
}
// Add adds a reader to be polled.
func (p *Poller) Add(reader *Sock) error {
rc := C.zpoller_add(p.zpollerT, unsafe.Pointer(reader.zsockT))
if int(rc) == -1 {
return fmt.Errorf("error adding reader")
}
p.socks = append(p.socks, reader)
return nil
}
// Remove removes a Sock from the poller
func (p *Poller) Remove(reader *Sock) {
numItems := len(p.socks)
for i := 0; i < numItems; i++ {
if p.socks[i] == reader {
if i == numItems-1 {
p.socks = p.socks[:i]
} else {
p.socks = append(p.socks[:i], p.socks[i+1:]...)
}
}
}
}
// Wait waits for the timeout period in milliseconds for a Pollin
// event, and returns the first socket that returns one
func (p *Poller) Wait(millis int) *Sock {
if p.zpollerT == nil {
// Null pointer. Something is wrong or we've already had `Destroy` invoked on us.
panic(ErrWaitAfterDestroy)
}
s := C.zpoller_wait(p.zpollerT, C.int(millis))
s = unsafe.Pointer(s)
if s == nil {
return nil
}
for _, sock := range p.socks {
if unsafe.Pointer(sock.zsockT) == s {
return sock
}
}
panic(fmt.Sprintf(
"Could not match received pointer with %v with any socket (%v)",
s, p.socks))
}
// Destroy destroys the Poller
func (p *Poller) Destroy() {
C.zpoller_destroy(&p.zpollerT)
}

View File

@ -1,171 +0,0 @@
package goczmq
/*
#include "czmq.h"
zactor_t *Zproxy_new () {
zactor_t *proxy = zactor_new(zproxy, NULL);
return proxy;
}
*/
import "C"
import (
"unsafe"
)
// Proxy wraps the CZMQ zproxy actor. A proxy actor switches
// messages between a frontend and backend socket, and also
// provides an optional capture socket messages can be
// mirrored to. The proxy can be paused and resumed.
type Proxy struct {
zactorT *C.struct__zactor_t
}
// NewProxy creates a new Proxy instance.
func NewProxy() *Proxy {
p := &Proxy{}
p.zactorT = C.Zproxy_new()
return p
}
// SetFrontend accepts a socket type and endpoint, and sends a message
// to the zactor thread telling it to set up a socket bound to the endpoint.
func (p *Proxy) SetFrontend(sockType int, endpoint string) error {
typeString := getStringType(sockType)
cmd := C.CString("FRONTEND")
defer C.free(unsafe.Pointer(cmd))
cTypeString := C.CString(typeString)
defer C.free(unsafe.Pointer(cTypeString))
cEndpoint := C.CString(endpoint)
defer C.free(unsafe.Pointer(cEndpoint))
rc := C.zstr_sendm(unsafe.Pointer(p.zactorT), cmd)
if rc == -1 {
return ErrActorCmd
}
rc = C.zstr_sendm(unsafe.Pointer(p.zactorT), cTypeString)
if rc == -1 {
return ErrActorCmd
}
rc = C.zstr_send(unsafe.Pointer(p.zactorT), cEndpoint)
if rc == -1 {
return ErrActorCmd
}
rc = C.zsock_wait(unsafe.Pointer(p.zactorT))
if rc == -1 {
return ErrActorCmd
}
return nil
}
// SetBackend accepts a socket type and endpoint, and sends a message
// to the zactor thread telling it to set up a socket bound to the endpoint.
func (p *Proxy) SetBackend(sockType int, endpoint string) error {
typeString := getStringType(sockType)
cmd := C.CString("BACKEND")
defer C.free(unsafe.Pointer(cmd))
cTypeString := C.CString(typeString)
defer C.free(unsafe.Pointer(cTypeString))
cEndpoint := C.CString(endpoint)
defer C.free(unsafe.Pointer(cEndpoint))
rc := C.zstr_sendm(unsafe.Pointer(p.zactorT), cmd)
if rc == -1 {
return ErrActorCmd
}
rc = C.zstr_sendm(unsafe.Pointer(p.zactorT), cTypeString)
if rc == -1 {
return ErrActorCmd
}
rc = C.zstr_send(unsafe.Pointer(p.zactorT), cEndpoint)
if rc == -1 {
return ErrActorCmd
}
rc = C.zsock_wait(unsafe.Pointer(p.zactorT))
if rc == -1 {
return ErrActorCmd
}
return nil
}
// SetCapture accepts a socket endpoint and sets up a Push socket bound
// to that endpoint, that sends a copy of all messages passing through
// the proxy.
func (p *Proxy) SetCapture(endpoint string) error {
cmd := C.CString("CAPTURE")
defer C.free(unsafe.Pointer(cmd))
cEndpoint := C.CString(endpoint)
defer C.free(unsafe.Pointer(cEndpoint))
rc := C.zstr_sendm(unsafe.Pointer(p.zactorT), cmd)
if rc == -1 {
return ErrActorCmd
}
rc = C.zstr_send(unsafe.Pointer(p.zactorT), cEndpoint)
if rc == -1 {
return ErrActorCmd
}
return nil
}
// Pause sends a message to the zproxy actor telling it to pause.
func (p *Proxy) Pause() error {
cmd := C.CString("PAUSE")
defer C.free(unsafe.Pointer(cmd))
rc := C.zstr_send(unsafe.Pointer(p.zactorT), cmd)
if rc == -1 {
return ErrActorCmd
}
return nil
}
// Resume sends a message to the zproxy actor telling it to resume.
func (p *Proxy) Resume() error {
cmd := C.CString("RESUME")
defer C.free(unsafe.Pointer(cmd))
rc := C.zstr_send(unsafe.Pointer(p.zactorT), cmd)
if rc == -1 {
return ErrActorCmd
}
return nil
}
// Verbose sets the proxy to log information to stdout.
func (p *Proxy) Verbose() error {
cmd := C.CString("VERBOSE")
defer C.free(unsafe.Pointer(cmd))
rc := C.zstr_send(unsafe.Pointer(p.zactorT), cmd)
if rc == -1 {
return ErrActorCmd
}
return nil
}
// Destroy destroys the proxy.
func (p *Proxy) Destroy() {
C.zactor_destroy(&p.zactorT)
}

View File

@ -1,120 +0,0 @@
package goczmq
import (
"C"
"io"
)
// ReadWriter provides an io.ReadWriter compatible
// interface for goczmq.Sock
type ReadWriter struct {
sock *Sock
poller *Poller
clientIDs []string
frame []byte
currentIndex int
timeoutMillis int
}
// NewReadWriter accepts a sock and returns a goczmq.ReadWriter. The
// io.ReadWriter should now be considered responsible for this
// Sock.
func NewReadWriter(sock *Sock) (*ReadWriter, error) {
rw := &ReadWriter{
sock: sock,
timeoutMillis: -1,
}
var err error
rw.poller, err = NewPoller(rw.sock)
return rw, err
}
// SetTimeout sets the timeout on Read in millisecond. If no new
// data is received within the timeout period, Read will return
// an ErrTimeout
func (r *ReadWriter) SetTimeout(ms int) {
r.timeoutMillis = ms
}
// Read satisifies io.Read
func (r *ReadWriter) Read(p []byte) (int, error) {
var totalRead int
var totalFrame int
var flag int
var err error
if r.currentIndex == 0 {
s := r.poller.Wait(r.timeoutMillis)
if s == nil {
return totalRead, ErrTimeout
}
r.frame, flag, err = s.RecvFrame()
if s.GetType() == Router && r.currentIndex == 0 {
r.clientIDs = append(r.clientIDs, string(r.frame))
r.frame, flag, err = s.RecvFrame()
}
if flag == FlagMore {
return totalRead, ErrMultiPartUnsupported
}
if err != nil {
return totalRead, io.EOF
}
}
totalRead += copy(p[:], r.frame[r.currentIndex:])
totalFrame += len(r.frame)
if totalFrame-r.currentIndex > len(p) {
r.currentIndex = totalRead
} else {
r.currentIndex = 0
err = io.EOF
}
return totalRead, err
}
// Write satisfies io.Write
func (r *ReadWriter) Write(p []byte) (int, error) {
var total int
if r.sock.GetType() == Router {
err := r.sock.SendFrame(r.GetLastClientID(), FlagMore)
if err != nil {
return total, err
}
}
err := r.sock.SendFrame(p, 0)
if err != nil {
return total, err
}
return len(p), nil
}
// GetLastClientID returns the id of the last client you received
// a message from if the underlying socket is a Router socket
func (r *ReadWriter) GetLastClientID() []byte {
id := []byte(r.clientIDs[0])
r.clientIDs = r.clientIDs[1:]
return id
}
// SetLastClientID lets you manually set the id of the client
// you last received a message from if the underlying socket
// is a Router socket
func (r *ReadWriter) SetLastClientID(id []byte) {
r.clientIDs = append(r.clientIDs, string(id))
}
// Destroy destroys both the ReadWriter and the underlying Sock
func (r *ReadWriter) Destroy() {
r.sock.Destroy()
r.poller.Destroy()
}

View File

@ -1,435 +0,0 @@
package goczmq
/*
#include "czmq.h"
#include <stdlib.h>
#include <string.h>
int Sock_connect(zsock_t *self, const char *format) {return zsock_connect(self, format, NULL);}
int Sock_disconnect(zsock_t *self, const char *format) {return zsock_disconnect(self, format, NULL);}
int Sock_bind(zsock_t *self, const char *format) {return zsock_bind(self, format, NULL);}
int Sock_unbind(zsock_t *self, const char *format) {return zsock_unbind(self, format, NULL);}
int Sock_sendframe(zsock_t *sock, const void *data, size_t size, int flags) {
zframe_t *frame = zframe_new (data, size);
int rc = zframe_send (&frame, sock, flags);
return rc;
}
*/
import "C"
import (
"os"
"runtime"
"strings"
"unsafe"
)
// Sock wraps the CZMQ zsock class.
type Sock struct {
zsockT *C.struct__zsock_t
file string
line int
zType int
clientIDs []string
}
func init() {
if err := os.Setenv("ZSYS_SIGHANDLER", "false"); err != nil {
panic(err)
}
}
// GetLastClientID returns the id of the last client you received
// a message from if the underlying socket is a Router socket
// DEPRECATED: See goczmq.ReadWriter
func (s *Sock) GetLastClientID() []byte {
id := []byte(s.clientIDs[0])
s.clientIDs = s.clientIDs[1:]
return id
}
// SetLastClientID lets you manually set the id of the client
// you last received a message from if the underlying socket
// is a Router socket
// DEPRECATED: See goczmq.ReadWriter
func (s *Sock) SetLastClientID(id []byte) {
s.clientIDs = append(s.clientIDs, string(id))
}
// NewSock creates a new socket. The caller source and
// line number are passed so CZMQ can report socket leaks
// intelligently.
func NewSock(t int) *Sock {
var s *Sock
_, file, line, ok := runtime.Caller(1)
if ok {
s = &Sock{
file: file,
line: line,
zType: t,
clientIDs: make([]string, 0),
}
} else {
s = &Sock{
file: "",
line: 0,
zType: t,
clientIDs: make([]string, 0),
}
}
cFile := C.CString(s.file)
defer C.free(unsafe.Pointer(cFile))
s.zsockT = C.zsock_new_checked(C.int(s.zType), cFile, C.size_t(s.line))
return s
}
// Connect connects a socket to an endpoint
// returns an error if the connect failed.
func (s *Sock) Connect(endpoint string) error {
cEndpoint := C.CString(endpoint)
defer C.free(unsafe.Pointer(cEndpoint))
rc := C.Sock_connect(s.zsockT, cEndpoint)
if rc != C.int(0) {
return ErrConnect
}
return nil
}
// Disconnect disconnects a socket from an endpoint. If returns
// an error if the endpoint was not found
func (s *Sock) Disconnect(endpoint string) error {
cEndpoint := C.CString(endpoint)
defer C.free(unsafe.Pointer(cEndpoint))
rc := C.Sock_disconnect(s.zsockT, cEndpoint)
if int(rc) == -1 {
return ErrDisconnect
}
return nil
}
// Bind binds a socket to an endpoint. On success returns
// the port number used for tcp transports, or 0 for other
// transports. On failure returns a -1 for port, and an error.
func (s *Sock) Bind(endpoint string) (int, error) {
cEndpoint := C.CString(endpoint)
defer C.free(unsafe.Pointer(cEndpoint))
port := C.Sock_bind(s.zsockT, cEndpoint)
if port == C.int(-1) {
return -1, ErrBind
}
return int(port), nil
}
// Unbind unbinds a socket from an endpoint. If returns
// an error if the endpoint was not found
func (s *Sock) Unbind(endpoint string) error {
cEndpoint := C.CString(endpoint)
defer C.free(unsafe.Pointer(cEndpoint))
rc := C.Sock_unbind(s.zsockT, cEndpoint)
if int(rc) == -1 {
return ErrUnbind
}
return nil
}
// Attach attaches a socket to zero or more endpoints. If endpoints is not null,
// parses as list of ZeroMQ endpoints, separated by commas, and prefixed by
// '@' (to bind the socket) or '>' (to attach the socket). If the endpoint
// does not start with '@' or '>', the serverish argument determines whether
// it is used to bind (serverish = true) or connect (serverish = false)
func (s *Sock) Attach(endpoints string, serverish bool) error {
cEndpoints := C.CString(endpoints)
defer C.free(unsafe.Pointer(cEndpoints))
rc := C.zsock_attach(s.zsockT, cEndpoints, C._Bool(serverish))
if rc == -1 {
return ErrSockAttach
}
return nil
}
// NewPub creates a Pub socket and calls Attach.
// The socket will Bind by default.
func NewPub(endpoints string) (*Sock, error) {
s := NewSock(Pub)
return s, s.Attach(endpoints, true)
}
// NewSub creates a Sub socket and calls Attach.
// 'subscribe' is a comma delimited list of topics to subscribe to.
// The socket will Connect by default.
func NewSub(endpoints string, subscribe string) (*Sock, error) {
s := NewSock(Sub)
subscriptions := strings.Split(subscribe, ",")
for _, topic := range subscriptions {
s.SetSubscribe(topic)
}
return s, s.Attach(endpoints, false)
}
// NewRep creates a Rep socket and calls Attach.
// The socket will Bind by default.
func NewRep(endpoints string) (*Sock, error) {
s := NewSock(Rep)
return s, s.Attach(endpoints, true)
}
// NewReq creates a Req socket and calls Attach.
// The socket will Connect by default.
func NewReq(endpoints string) (*Sock, error) {
s := NewSock(Req)
return s, s.Attach(endpoints, false)
}
// NewPull creates a Pull socket and calls Attach.
// The socket will Bind by default.
func NewPull(endpoints string) (*Sock, error) {
s := NewSock(Pull)
return s, s.Attach(endpoints, true)
}
// NewPush creates a Push socket and calls Attach.
// The socket will Connect by default.
func NewPush(endpoints string) (*Sock, error) {
s := NewSock(Push)
return s, s.Attach(endpoints, false)
}
// NewRouter creates a Router socket and calls Attach.
// The socket will Bind by default.
func NewRouter(endpoints string) (*Sock, error) {
s := NewSock(Router)
return s, s.Attach(endpoints, true)
}
// NewDealer creates a Dealer socket and calls Attach.
// The socket will Connect by default.
func NewDealer(endpoints string) (*Sock, error) {
s := NewSock(Dealer)
return s, s.Attach(endpoints, false)
}
// NewXPub creates an XPub socket and calls Attach.
// The socket will Bind by default.
func NewXPub(endpoints string) (*Sock, error) {
s := NewSock(XPub)
return s, s.Attach(endpoints, true)
}
// NewXSub creates an XSub socket and calls Attach.
// The socket will Connect by default.
func NewXSub(endpoints string) (*Sock, error) {
s := NewSock(XSub)
return s, s.Attach(endpoints, false)
}
// NewPair creates a Pair socket and calls Attach.
// The socket will Connect by default.
func NewPair(endpoints string) (*Sock, error) {
s := NewSock(Pair)
return s, s.Attach(endpoints, false)
}
// NewStream creates a Stream socket and calls Attach.
// The socket will Connect by default.
func NewStream(endpoints string) (*Sock, error) {
s := NewSock(Stream)
return s, s.Attach(endpoints, false)
}
// Pollin returns true if there is a Pollin
// event on the socket
func (s *Sock) Pollin() bool {
return s.Events() == Pollin
}
// Pollout returns true if there is a Pollout
// event on the socket
func (s *Sock) Pollout() bool {
return s.Events() == Pollout
}
// SendFrame sends a byte array via the socket. For the flags
// value, use 0 for a single message, or SNDFlagMore if it is
// a multi-part message
func (s *Sock) SendFrame(data []byte, flags int) error {
var rc C.int
if len(data) == 0 {
rc = C.Sock_sendframe(s.zsockT, nil, C.size_t(0), C.int(flags))
} else {
rc = C.Sock_sendframe(s.zsockT, unsafe.Pointer(&data[0]), C.size_t(len(data)), C.int(flags))
}
if rc == C.int(-1) {
return ErrSendFrame
}
return nil
}
// RecvFrame reads a frame from the socket and returns it
// as a byte array, along with a more flag and and error
// (if there is an error)
func (s *Sock) RecvFrame() ([]byte, int, error) {
if s.zsockT == nil {
return nil, -1, ErrRecvFrameAfterDestroy
}
frame := C.zframe_recv(unsafe.Pointer(s.zsockT))
if frame == nil {
return []byte{0}, 0, ErrRecvFrame
}
dataSize := C.zframe_size(frame)
dataPtr := C.zframe_data(frame)
b := C.GoBytes(unsafe.Pointer(dataPtr), C.int(dataSize))
more := C.zframe_more(frame)
C.zframe_destroy(&frame)
return b, int(more), nil
}
// RecvFrameNoWait receives a frame from the socket
// and returns it as a byte array if one is waiting.
// Returns an empty frame, a 0 more flag and an error
// if one is not immediately available
func (s *Sock) RecvFrameNoWait() ([]byte, int, error) {
if !s.Pollin() {
return []byte{0}, 0, ErrRecvFrame
}
return s.RecvFrame()
}
// SendMessage accepts an array of byte arrays and
// sends it as a multi-part message.
func (s *Sock) SendMessage(parts [][]byte) error {
var f int
numParts := len(parts)
for i, val := range parts {
if i == numParts-1 {
f = 0
} else {
f = FlagMore
}
err := s.SendFrame(val, f)
if err != nil {
return err
}
}
return nil
}
// RecvMessage receives a full message from the socket
// and returns it as an array of byte arrays.
func (s *Sock) RecvMessage() ([][]byte, error) {
var msg [][]byte
for {
frame, flag, err := s.RecvFrame()
if err != nil {
return msg, err
}
msg = append(msg, frame)
if flag != FlagMore {
break
}
}
return msg, nil
}
// Read provides an io.Reader interface to a zeromq socket
// DEPRECATED: see goczmq.ReadWriter
func (s *Sock) Read(p []byte) (int, error) {
var totalRead int
var totalFrame int
frame, flag, err := s.RecvFrame()
if err != nil {
return totalRead, err
}
if s.GetType() == Router {
s.clientIDs = append(s.clientIDs, string(frame))
} else {
totalRead += copy(p[:], frame[:])
totalFrame += len(frame)
}
for flag == FlagMore {
frame, flag, err = s.RecvFrame()
if err != nil {
return totalRead, err
}
totalRead += copy(p[totalRead:], frame[:])
totalFrame += len(frame)
}
if totalFrame > len(p) {
err = ErrSliceFull
} else {
err = nil
}
return totalRead, err
}
// Write provides an io.Writer interface to a zeromq socket
// DEPRECATED: See goczmq.ReadWriter
func (s *Sock) Write(p []byte) (int, error) {
var total int
if s.GetType() == Router {
err := s.SendFrame(s.GetLastClientID(), FlagMore)
if err != nil {
return total, err
}
}
err := s.SendFrame(p, 0)
if err != nil {
return total, err
}
return len(p), nil
}
// RecvMessageNoWait receives a full message from the socket
// and returns it as an array of byte arrays if one is waiting.
// Returns an empty message and an error if one is not immediately
// available
func (s *Sock) RecvMessageNoWait() ([][]byte, error) {
var msg [][]byte
if !s.Pollin() {
return msg, ErrRecvMessage
}
for {
frame, flag, err := s.RecvFrame()
if err != nil {
return msg, err
}
msg = append(msg, frame)
if flag != FlagMore {
break
}
}
return msg, nil
}
// GetType returns the socket's type
func (s *Sock) GetType() int {
return s.zType
}
// Destroy destroys the underlying zsockT.
func (s *Sock) Destroy() {
cFile := C.CString(s.file)
defer C.free(unsafe.Pointer(cFile))
C.zsock_destroy_checked(&s.zsockT, cFile, C.size_t(s.line))
}

View File

@ -1,30 +0,0 @@
// +build draft
package goczmq
/*
#include "czmq.h"
*/
import "C"
const (
// Scatter is a ZMQ_SCATTER socket type
Scatter = int(C.ZMQ_SCATTER)
// Gather is a ZMQ_GATHER socket type
Gather = int(C.ZMQ_GATHER)
)
// NewGather creates a Gather socket and calls Attach.
// The socket will Bind by default.
func NewGather(endpoints string) (*Sock, error) {
s := NewSock(Gather)
return s, s.Attach(endpoints, true)
}
// NewScatter creates a Scatter socket and calls Attach.
// The socket will Connect by default.
func NewScatter(endpoints string) (*Sock, error) {
s := NewSock(Scatter)
return s, s.Attach(endpoints, false)
}

View File

@ -1,765 +0,0 @@
//go:generate gsl sockopts.xml
package goczmq
/* =========================================================================
zsock_option - get/set 0MQ socket options
****************************************************
* GENERATED SOURCE CODE, DO NOT EDIT!! *
* TO CHANGE THIS, EDIT sockopts.gsl *
* AND RUN gsl -q sockopts.xml *
****************************************************
Copyright (c) the Contributors as noted in the AUTHORS file.
This file is part of goczmq, the high-level go binding for CZMQ:
http://github.com/zeromq/goczmq
This Source Code Form is subject to the terms of the Mozilla Public
License, v. 2.0. If a copy of the MPL was not distributed with this
file, You can obtain one at http://mozilla.org/MPL/2.0/.
=========================================================================
*/
/*
#include "czmq.h"
#include <stdlib.h>
#include <string.h>
*/
import "C"
import (
"unsafe"
)
// SetHeartbeatIvl sets the heartbeat_ivl option for the socket
func (s *Sock) SetHeartbeatIvl(val int) {
C.zsock_set_heartbeat_ivl(unsafe.Pointer(s.zsockT), C.int(val))
}
// HeartbeatIvl returns the current value of the socket's heartbeat_ivl option
func (s *Sock) HeartbeatIvl() int {
val := C.zsock_heartbeat_ivl(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetHeartbeatTtl sets the heartbeat_ttl option for the socket
func (s *Sock) SetHeartbeatTtl(val int) {
C.zsock_set_heartbeat_ttl(unsafe.Pointer(s.zsockT), C.int(val))
}
// HeartbeatTtl returns the current value of the socket's heartbeat_ttl option
func (s *Sock) HeartbeatTtl() int {
val := C.zsock_heartbeat_ttl(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetHeartbeatTimeout sets the heartbeat_timeout option for the socket
func (s *Sock) SetHeartbeatTimeout(val int) {
C.zsock_set_heartbeat_timeout(unsafe.Pointer(s.zsockT), C.int(val))
}
// HeartbeatTimeout returns the current value of the socket's heartbeat_timeout option
func (s *Sock) HeartbeatTimeout() int {
val := C.zsock_heartbeat_timeout(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetUseFd sets the use_fd option for the socket
func (s *Sock) SetUseFd(val int) {
C.zsock_set_use_fd(unsafe.Pointer(s.zsockT), C.int(val))
}
// UseFd returns the current value of the socket's use_fd option
func (s *Sock) UseFd() int {
val := C.zsock_use_fd(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetXPubManual sets the xpub_manual option for the socket
func (s *Sock) SetXPubManual(val int) {
C.zsock_set_xpub_manual(unsafe.Pointer(s.zsockT), C.int(val))
}
// SetXPubWelcomeMsg sets the xpub_welcome_msg option for the socket
func (s *Sock) SetXPubWelcomeMsg(val string) {
cVal := C.CString(val)
defer C.free(unsafe.Pointer(cVal))
C.zsock_set_xpub_welcome_msg(unsafe.Pointer(s.zsockT), cVal)
}
// SetStreamNotify sets the stream_notify option for the socket
func (s *Sock) SetStreamNotify(val int) {
C.zsock_set_stream_notify(unsafe.Pointer(s.zsockT), C.int(val))
}
// SetInvertMatching sets the invert_matching option for the socket
func (s *Sock) SetInvertMatching(val int) {
C.zsock_set_invert_matching(unsafe.Pointer(s.zsockT), C.int(val))
}
// InvertMatching returns the current value of the socket's invert_matching option
func (s *Sock) InvertMatching() int {
val := C.zsock_invert_matching(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetXPubVerboser sets the xpub_verboser option for the socket
func (s *Sock) SetXPubVerboser(val int) {
C.zsock_set_xpub_verboser(unsafe.Pointer(s.zsockT), C.int(val))
}
// SetConnectTimeout sets the connect_timeout option for the socket
func (s *Sock) SetConnectTimeout(val int) {
C.zsock_set_connect_timeout(unsafe.Pointer(s.zsockT), C.int(val))
}
// ConnectTimeout returns the current value of the socket's connect_timeout option
func (s *Sock) ConnectTimeout() int {
val := C.zsock_connect_timeout(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetTcpMaxrt sets the tcp_maxrt option for the socket
func (s *Sock) SetTcpMaxrt(val int) {
C.zsock_set_tcp_maxrt(unsafe.Pointer(s.zsockT), C.int(val))
}
// TcpMaxrt returns the current value of the socket's tcp_maxrt option
func (s *Sock) TcpMaxrt() int {
val := C.zsock_tcp_maxrt(unsafe.Pointer(s.zsockT))
return int(val)
}
// ThreadSafe returns the current value of the socket's thread_safe option
func (s *Sock) ThreadSafe() int {
val := C.zsock_thread_safe(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetMulticastMaxtpdu sets the multicast_maxtpdu option for the socket
func (s *Sock) SetMulticastMaxtpdu(val int) {
C.zsock_set_multicast_maxtpdu(unsafe.Pointer(s.zsockT), C.int(val))
}
// MulticastMaxtpdu returns the current value of the socket's multicast_maxtpdu option
func (s *Sock) MulticastMaxtpdu() int {
val := C.zsock_multicast_maxtpdu(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetVmciBufferSize sets the vmci_buffer_size option for the socket
func (s *Sock) SetVmciBufferSize(val int) {
C.zsock_set_vmci_buffer_size(unsafe.Pointer(s.zsockT), C.int(val))
}
// VmciBufferSize returns the current value of the socket's vmci_buffer_size option
func (s *Sock) VmciBufferSize() int {
val := C.zsock_vmci_buffer_size(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetVmciBufferMinSize sets the vmci_buffer_min_size option for the socket
func (s *Sock) SetVmciBufferMinSize(val int) {
C.zsock_set_vmci_buffer_min_size(unsafe.Pointer(s.zsockT), C.int(val))
}
// VmciBufferMinSize returns the current value of the socket's vmci_buffer_min_size option
func (s *Sock) VmciBufferMinSize() int {
val := C.zsock_vmci_buffer_min_size(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetVmciBufferMaxSize sets the vmci_buffer_max_size option for the socket
func (s *Sock) SetVmciBufferMaxSize(val int) {
C.zsock_set_vmci_buffer_max_size(unsafe.Pointer(s.zsockT), C.int(val))
}
// VmciBufferMaxSize returns the current value of the socket's vmci_buffer_max_size option
func (s *Sock) VmciBufferMaxSize() int {
val := C.zsock_vmci_buffer_max_size(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetVmciConnectTimeout sets the vmci_connect_timeout option for the socket
func (s *Sock) SetVmciConnectTimeout(val int) {
C.zsock_set_vmci_connect_timeout(unsafe.Pointer(s.zsockT), C.int(val))
}
// VmciConnectTimeout returns the current value of the socket's vmci_connect_timeout option
func (s *Sock) VmciConnectTimeout() int {
val := C.zsock_vmci_connect_timeout(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetTos sets the tos option for the socket
func (s *Sock) SetTos(val int) {
C.zsock_set_tos(unsafe.Pointer(s.zsockT), C.int(val))
}
// Tos returns the current value of the socket's tos option
func (s *Sock) Tos() int {
val := C.zsock_tos(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetRouterHandover sets the router_handover option for the socket
func (s *Sock) SetRouterHandover(val int) {
C.zsock_set_router_handover(unsafe.Pointer(s.zsockT), C.int(val))
}
// SetConnectRid sets the connect_rid option for the socket
func (s *Sock) SetConnectRid(val string) {
cVal := C.CString(val)
defer C.free(unsafe.Pointer(cVal))
C.zsock_set_connect_rid(unsafe.Pointer(s.zsockT), cVal)
}
// SetHandshakeIvl sets the handshake_ivl option for the socket
func (s *Sock) SetHandshakeIvl(val int) {
C.zsock_set_handshake_ivl(unsafe.Pointer(s.zsockT), C.int(val))
}
// HandshakeIvl returns the current value of the socket's handshake_ivl option
func (s *Sock) HandshakeIvl() int {
val := C.zsock_handshake_ivl(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetSocksProxy sets the socks_proxy option for the socket
func (s *Sock) SetSocksProxy(val string) {
cVal := C.CString(val)
defer C.free(unsafe.Pointer(cVal))
C.zsock_set_socks_proxy(unsafe.Pointer(s.zsockT), cVal)
}
// SocksProxy returns the current value of the socket's socks_proxy option
func (s *Sock) SocksProxy() string {
val := C.zsock_socks_proxy(unsafe.Pointer(s.zsockT))
return C.GoString(val)
}
// SetXPubNodrop sets the xpub_nodrop option for the socket
func (s *Sock) SetXPubNodrop(val int) {
C.zsock_set_xpub_nodrop(unsafe.Pointer(s.zsockT), C.int(val))
}
// SetRouterMandatory sets the router_mandatory option for the socket
func (s *Sock) SetRouterMandatory(val int) {
C.zsock_set_router_mandatory(unsafe.Pointer(s.zsockT), C.int(val))
}
// SetProbeRouter sets the probe_router option for the socket
func (s *Sock) SetProbeRouter(val int) {
C.zsock_set_probe_router(unsafe.Pointer(s.zsockT), C.int(val))
}
// SetReqRelaxed sets the req_relaxed option for the socket
func (s *Sock) SetReqRelaxed(val int) {
C.zsock_set_req_relaxed(unsafe.Pointer(s.zsockT), C.int(val))
}
// SetReqCorrelate sets the req_correlate option for the socket
func (s *Sock) SetReqCorrelate(val int) {
C.zsock_set_req_correlate(unsafe.Pointer(s.zsockT), C.int(val))
}
// SetConflate sets the conflate option for the socket
func (s *Sock) SetConflate(val int) {
C.zsock_set_conflate(unsafe.Pointer(s.zsockT), C.int(val))
}
// SetZapDomain sets the zap_domain option for the socket
func (s *Sock) SetZapDomain(val string) {
cVal := C.CString(val)
defer C.free(unsafe.Pointer(cVal))
C.zsock_set_zap_domain(unsafe.Pointer(s.zsockT), cVal)
}
// ZapDomain returns the current value of the socket's zap_domain option
func (s *Sock) ZapDomain() string {
val := C.zsock_zap_domain(unsafe.Pointer(s.zsockT))
return C.GoString(val)
}
// Mechanism returns the current value of the socket's mechanism option
func (s *Sock) Mechanism() int {
val := C.zsock_mechanism(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetPlainServer sets the plain_server option for the socket
func (s *Sock) SetPlainServer(val int) {
C.zsock_set_plain_server(unsafe.Pointer(s.zsockT), C.int(val))
}
// PlainServer returns the current value of the socket's plain_server option
func (s *Sock) PlainServer() int {
val := C.zsock_plain_server(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetPlainUsername sets the plain_username option for the socket
func (s *Sock) SetPlainUsername(val string) {
cVal := C.CString(val)
defer C.free(unsafe.Pointer(cVal))
C.zsock_set_plain_username(unsafe.Pointer(s.zsockT), cVal)
}
// PlainUsername returns the current value of the socket's plain_username option
func (s *Sock) PlainUsername() string {
val := C.zsock_plain_username(unsafe.Pointer(s.zsockT))
return C.GoString(val)
}
// SetPlainPassword sets the plain_password option for the socket
func (s *Sock) SetPlainPassword(val string) {
cVal := C.CString(val)
defer C.free(unsafe.Pointer(cVal))
C.zsock_set_plain_password(unsafe.Pointer(s.zsockT), cVal)
}
// PlainPassword returns the current value of the socket's plain_password option
func (s *Sock) PlainPassword() string {
val := C.zsock_plain_password(unsafe.Pointer(s.zsockT))
return C.GoString(val)
}
// SetCurveServer sets the curve_server option for the socket
func (s *Sock) SetCurveServer(val int) {
C.zsock_set_curve_server(unsafe.Pointer(s.zsockT), C.int(val))
}
// CurveServer returns the current value of the socket's curve_server option
func (s *Sock) CurveServer() int {
val := C.zsock_curve_server(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetCurvePublickey sets the curve_publickey option for the socket
func (s *Sock) SetCurvePublickey(val string) {
cVal := C.CString(val)
defer C.free(unsafe.Pointer(cVal))
C.zsock_set_curve_publickey(unsafe.Pointer(s.zsockT), cVal)
}
// CurvePublickey returns the current value of the socket's curve_publickey option
func (s *Sock) CurvePublickey() string {
val := C.zsock_curve_publickey(unsafe.Pointer(s.zsockT))
return C.GoString(val)
}
// SetCurveSecretkey sets the curve_secretkey option for the socket
func (s *Sock) SetCurveSecretkey(val string) {
cVal := C.CString(val)
defer C.free(unsafe.Pointer(cVal))
C.zsock_set_curve_secretkey(unsafe.Pointer(s.zsockT), cVal)
}
// CurveSecretkey returns the current value of the socket's curve_secretkey option
func (s *Sock) CurveSecretkey() string {
val := C.zsock_curve_secretkey(unsafe.Pointer(s.zsockT))
return C.GoString(val)
}
// SetCurveServerkey sets the curve_serverkey option for the socket
func (s *Sock) SetCurveServerkey(val string) {
cVal := C.CString(val)
defer C.free(unsafe.Pointer(cVal))
C.zsock_set_curve_serverkey(unsafe.Pointer(s.zsockT), cVal)
}
// CurveServerkey returns the current value of the socket's curve_serverkey option
func (s *Sock) CurveServerkey() string {
val := C.zsock_curve_serverkey(unsafe.Pointer(s.zsockT))
return C.GoString(val)
}
// SetGssapiServer sets the gssapi_server option for the socket
func (s *Sock) SetGssapiServer(val int) {
C.zsock_set_gssapi_server(unsafe.Pointer(s.zsockT), C.int(val))
}
// GssapiServer returns the current value of the socket's gssapi_server option
func (s *Sock) GssapiServer() int {
val := C.zsock_gssapi_server(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetGssapiPlaintext sets the gssapi_plaintext option for the socket
func (s *Sock) SetGssapiPlaintext(val int) {
C.zsock_set_gssapi_plaintext(unsafe.Pointer(s.zsockT), C.int(val))
}
// GssapiPlaintext returns the current value of the socket's gssapi_plaintext option
func (s *Sock) GssapiPlaintext() int {
val := C.zsock_gssapi_plaintext(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetGssapiPrincipal sets the gssapi_principal option for the socket
func (s *Sock) SetGssapiPrincipal(val string) {
cVal := C.CString(val)
defer C.free(unsafe.Pointer(cVal))
C.zsock_set_gssapi_principal(unsafe.Pointer(s.zsockT), cVal)
}
// GssapiPrincipal returns the current value of the socket's gssapi_principal option
func (s *Sock) GssapiPrincipal() string {
val := C.zsock_gssapi_principal(unsafe.Pointer(s.zsockT))
return C.GoString(val)
}
// SetGssapiServicePrincipal sets the gssapi_service_principal option for the socket
func (s *Sock) SetGssapiServicePrincipal(val string) {
cVal := C.CString(val)
defer C.free(unsafe.Pointer(cVal))
C.zsock_set_gssapi_service_principal(unsafe.Pointer(s.zsockT), cVal)
}
// GssapiServicePrincipal returns the current value of the socket's gssapi_service_principal option
func (s *Sock) GssapiServicePrincipal() string {
val := C.zsock_gssapi_service_principal(unsafe.Pointer(s.zsockT))
return C.GoString(val)
}
// SetIpv6 sets the ipv6 option for the socket
func (s *Sock) SetIpv6(val int) {
C.zsock_set_ipv6(unsafe.Pointer(s.zsockT), C.int(val))
}
// Ipv6 returns the current value of the socket's ipv6 option
func (s *Sock) Ipv6() int {
val := C.zsock_ipv6(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetImmediate sets the immediate option for the socket
func (s *Sock) SetImmediate(val int) {
C.zsock_set_immediate(unsafe.Pointer(s.zsockT), C.int(val))
}
// Immediate returns the current value of the socket's immediate option
func (s *Sock) Immediate() int {
val := C.zsock_immediate(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetRouterRaw sets the router_raw option for the socket
func (s *Sock) SetRouterRaw(val int) {
C.zsock_set_router_raw(unsafe.Pointer(s.zsockT), C.int(val))
}
// SetIpv4only sets the ipv4only option for the socket
func (s *Sock) SetIpv4only(val int) {
C.zsock_set_ipv4only(unsafe.Pointer(s.zsockT), C.int(val))
}
// Ipv4only returns the current value of the socket's ipv4only option
func (s *Sock) Ipv4only() int {
val := C.zsock_ipv4only(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetDelayAttachOnConnect sets the delay_attach_on_connect option for the socket
func (s *Sock) SetDelayAttachOnConnect(val int) {
C.zsock_set_delay_attach_on_connect(unsafe.Pointer(s.zsockT), C.int(val))
}
// Type returns the current value of the socket's type option
func (s *Sock) Type() int {
val := C.zsock_type(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetSndhwm sets the sndhwm option for the socket
func (s *Sock) SetSndhwm(val int) {
C.zsock_set_sndhwm(unsafe.Pointer(s.zsockT), C.int(val))
}
// Sndhwm returns the current value of the socket's sndhwm option
func (s *Sock) Sndhwm() int {
val := C.zsock_sndhwm(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetRcvhwm sets the rcvhwm option for the socket
func (s *Sock) SetRcvhwm(val int) {
C.zsock_set_rcvhwm(unsafe.Pointer(s.zsockT), C.int(val))
}
// Rcvhwm returns the current value of the socket's rcvhwm option
func (s *Sock) Rcvhwm() int {
val := C.zsock_rcvhwm(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetAffinity sets the affinity option for the socket
func (s *Sock) SetAffinity(val int) {
C.zsock_set_affinity(unsafe.Pointer(s.zsockT), C.int(val))
}
// Affinity returns the current value of the socket's affinity option
func (s *Sock) Affinity() int {
val := C.zsock_affinity(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetSubscribe sets the subscribe option for the socket
func (s *Sock) SetSubscribe(val string) {
cVal := C.CString(val)
defer C.free(unsafe.Pointer(cVal))
C.zsock_set_subscribe(unsafe.Pointer(s.zsockT), cVal)
}
// SetUnsubscribe sets the unsubscribe option for the socket
func (s *Sock) SetUnsubscribe(val string) {
cVal := C.CString(val)
defer C.free(unsafe.Pointer(cVal))
C.zsock_set_unsubscribe(unsafe.Pointer(s.zsockT), cVal)
}
// SetIdentity sets the identity option for the socket
func (s *Sock) SetIdentity(val string) {
cVal := C.CString(val)
defer C.free(unsafe.Pointer(cVal))
C.zsock_set_identity(unsafe.Pointer(s.zsockT), cVal)
}
// Identity returns the current value of the socket's identity option
func (s *Sock) Identity() string {
val := C.zsock_identity(unsafe.Pointer(s.zsockT))
return C.GoString(val)
}
// SetRate sets the rate option for the socket
func (s *Sock) SetRate(val int) {
C.zsock_set_rate(unsafe.Pointer(s.zsockT), C.int(val))
}
// Rate returns the current value of the socket's rate option
func (s *Sock) Rate() int {
val := C.zsock_rate(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetRecoveryIvl sets the recovery_ivl option for the socket
func (s *Sock) SetRecoveryIvl(val int) {
C.zsock_set_recovery_ivl(unsafe.Pointer(s.zsockT), C.int(val))
}
// RecoveryIvl returns the current value of the socket's recovery_ivl option
func (s *Sock) RecoveryIvl() int {
val := C.zsock_recovery_ivl(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetSndbuf sets the sndbuf option for the socket
func (s *Sock) SetSndbuf(val int) {
C.zsock_set_sndbuf(unsafe.Pointer(s.zsockT), C.int(val))
}
// Sndbuf returns the current value of the socket's sndbuf option
func (s *Sock) Sndbuf() int {
val := C.zsock_sndbuf(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetRcvbuf sets the rcvbuf option for the socket
func (s *Sock) SetRcvbuf(val int) {
C.zsock_set_rcvbuf(unsafe.Pointer(s.zsockT), C.int(val))
}
// Rcvbuf returns the current value of the socket's rcvbuf option
func (s *Sock) Rcvbuf() int {
val := C.zsock_rcvbuf(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetLinger sets the linger option for the socket
func (s *Sock) SetLinger(val int) {
C.zsock_set_linger(unsafe.Pointer(s.zsockT), C.int(val))
}
// Linger returns the current value of the socket's linger option
func (s *Sock) Linger() int {
val := C.zsock_linger(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetReconnectIvl sets the reconnect_ivl option for the socket
func (s *Sock) SetReconnectIvl(val int) {
C.zsock_set_reconnect_ivl(unsafe.Pointer(s.zsockT), C.int(val))
}
// ReconnectIvl returns the current value of the socket's reconnect_ivl option
func (s *Sock) ReconnectIvl() int {
val := C.zsock_reconnect_ivl(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetReconnectIvlMax sets the reconnect_ivl_max option for the socket
func (s *Sock) SetReconnectIvlMax(val int) {
C.zsock_set_reconnect_ivl_max(unsafe.Pointer(s.zsockT), C.int(val))
}
// ReconnectIvlMax returns the current value of the socket's reconnect_ivl_max option
func (s *Sock) ReconnectIvlMax() int {
val := C.zsock_reconnect_ivl_max(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetBacklog sets the backlog option for the socket
func (s *Sock) SetBacklog(val int) {
C.zsock_set_backlog(unsafe.Pointer(s.zsockT), C.int(val))
}
// Backlog returns the current value of the socket's backlog option
func (s *Sock) Backlog() int {
val := C.zsock_backlog(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetMaxmsgsize sets the maxmsgsize option for the socket
func (s *Sock) SetMaxmsgsize(val int) {
C.zsock_set_maxmsgsize(unsafe.Pointer(s.zsockT), C.int(val))
}
// Maxmsgsize returns the current value of the socket's maxmsgsize option
func (s *Sock) Maxmsgsize() int {
val := C.zsock_maxmsgsize(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetMulticastHops sets the multicast_hops option for the socket
func (s *Sock) SetMulticastHops(val int) {
C.zsock_set_multicast_hops(unsafe.Pointer(s.zsockT), C.int(val))
}
// MulticastHops returns the current value of the socket's multicast_hops option
func (s *Sock) MulticastHops() int {
val := C.zsock_multicast_hops(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetRcvtimeo sets the rcvtimeo option for the socket
func (s *Sock) SetRcvtimeo(val int) {
C.zsock_set_rcvtimeo(unsafe.Pointer(s.zsockT), C.int(val))
}
// Rcvtimeo returns the current value of the socket's rcvtimeo option
func (s *Sock) Rcvtimeo() int {
val := C.zsock_rcvtimeo(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetSndtimeo sets the sndtimeo option for the socket
func (s *Sock) SetSndtimeo(val int) {
C.zsock_set_sndtimeo(unsafe.Pointer(s.zsockT), C.int(val))
}
// Sndtimeo returns the current value of the socket's sndtimeo option
func (s *Sock) Sndtimeo() int {
val := C.zsock_sndtimeo(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetXPubVerbose sets the xpub_verbose option for the socket
func (s *Sock) SetXPubVerbose(val int) {
C.zsock_set_xpub_verbose(unsafe.Pointer(s.zsockT), C.int(val))
}
// SetTcpKeepalive sets the tcp_keepalive option for the socket
func (s *Sock) SetTcpKeepalive(val int) {
C.zsock_set_tcp_keepalive(unsafe.Pointer(s.zsockT), C.int(val))
}
// TcpKeepalive returns the current value of the socket's tcp_keepalive option
func (s *Sock) TcpKeepalive() int {
val := C.zsock_tcp_keepalive(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetTcpKeepaliveIdle sets the tcp_keepalive_idle option for the socket
func (s *Sock) SetTcpKeepaliveIdle(val int) {
C.zsock_set_tcp_keepalive_idle(unsafe.Pointer(s.zsockT), C.int(val))
}
// TcpKeepaliveIdle returns the current value of the socket's tcp_keepalive_idle option
func (s *Sock) TcpKeepaliveIdle() int {
val := C.zsock_tcp_keepalive_idle(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetTcpKeepaliveCnt sets the tcp_keepalive_cnt option for the socket
func (s *Sock) SetTcpKeepaliveCnt(val int) {
C.zsock_set_tcp_keepalive_cnt(unsafe.Pointer(s.zsockT), C.int(val))
}
// TcpKeepaliveCnt returns the current value of the socket's tcp_keepalive_cnt option
func (s *Sock) TcpKeepaliveCnt() int {
val := C.zsock_tcp_keepalive_cnt(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetTcpKeepaliveIntvl sets the tcp_keepalive_intvl option for the socket
func (s *Sock) SetTcpKeepaliveIntvl(val int) {
C.zsock_set_tcp_keepalive_intvl(unsafe.Pointer(s.zsockT), C.int(val))
}
// TcpKeepaliveIntvl returns the current value of the socket's tcp_keepalive_intvl option
func (s *Sock) TcpKeepaliveIntvl() int {
val := C.zsock_tcp_keepalive_intvl(unsafe.Pointer(s.zsockT))
return int(val)
}
// SetTcpAcceptFilter sets the tcp_accept_filter option for the socket
func (s *Sock) SetTcpAcceptFilter(val string) {
cVal := C.CString(val)
defer C.free(unsafe.Pointer(cVal))
C.zsock_set_tcp_accept_filter(unsafe.Pointer(s.zsockT), cVal)
}
// TcpAcceptFilter returns the current value of the socket's tcp_accept_filter option
func (s *Sock) TcpAcceptFilter() string {
val := C.zsock_tcp_accept_filter(unsafe.Pointer(s.zsockT))
return C.GoString(val)
}
// Rcvmore returns the current value of the socket's rcvmore option
func (s *Sock) Rcvmore() int {
val := C.zsock_rcvmore(unsafe.Pointer(s.zsockT))
return int(val)
}
// Fd returns the current value of the socket's fd option
func (s *Sock) Fd() int {
val := C.zsock_fd(unsafe.Pointer(s.zsockT))
return int(val)
}
// Events returns the current value of the socket's events option
func (s *Sock) Events() int {
val := C.zsock_events(unsafe.Pointer(s.zsockT))
return int(val)
}
// LastEndpoint returns the current value of the socket's last_endpoint option
func (s *Sock) LastEndpoint() string {
val := C.zsock_last_endpoint(unsafe.Pointer(s.zsockT))
return C.GoString(val)
}

View File

@ -1,84 +0,0 @@
.# This is a code generator built using the iMatix GSL code generation
.# language. See https://github.com/imatix/gsl for details. This script
.# is licensed under MIT/X11.
.#
.output "./sock_option.go"
//go:generate gsl sockopts.xml
package goczmq
/* =========================================================================
zsock_option - get/set 0MQ socket options
****************************************************
* GENERATED SOURCE CODE, DO NOT EDIT!! *
* TO CHANGE THIS, EDIT sockopts.gsl *
* AND RUN gsl -q sockopts.xml *
****************************************************
Copyright (c) the Contributors as noted in the AUTHORS file.
This file is part of goczmq, the high-level go binding for CZMQ:
http://github.com/zeromq/goczmq
This Source Code Form is subject to the terms of the Mozilla Public
License, v. 2.0. If a copy of the MPL was not distributed with this
file, You can obtain one at http://mozilla.org/MPL/2.0/.
=========================================================================
*/
/*
#include "czmq.h"
#include <stdlib.h>
#include <string.h>
*/
import "C"
import (
"unsafe"
)
.for version
.if major = "4"
.for option
.if mode = "rw" | mode = "w"
.if type = "uint64" | type = "int64" | type = "uint32" | type = "int"
// Set$(name:pascal) sets the $(name) option for the socket
func (s *Sock) Set$(name:pascal)(val $(gotype)) {
C.zsock_set_$(name)(unsafe.Pointer(s.zsockT), C.int(val))
}
.endif
.if type = "string" | type = "key"
// Set$(name:pascal) sets the $(name) option for the socket
func (s *Sock) Set$(name:pascal)(val $(gotype)) {
cVal := C.CString(val)
defer C.free(unsafe.Pointer(cVal))
C.zsock_set_$(name)(unsafe.Pointer(s.zsockT), cVal)
}
.endif
.endif
.if mode = "rw" | mode = "r"
.if type = "uint64" | type = "int64" | type = "uint32" | type = "int"
// $(name:pascal) returns the current value of the socket's $(name) option
func (s *Sock) $(name:pascal)() $(gotype) {
val := C.zsock_$(name)(unsafe.Pointer(s.zsockT))
return int(val)
}
.endif
.if type = "string" | type = "key"
// $(name:pascal) returns the current value of the socket's $(name) option
func (s *Sock) $(name:pascal)() $(gotype) {
val := C.zsock_$(name)(unsafe.Pointer(s.zsockT))
return C.GoString(val)
}
.endif
.endif
.endfor
.endif
.for source
$(string.trim(.):)
.endfor
.endfor

View File

@ -1,83 +0,0 @@
.# This is a code generator built using the iMatix GSL code generation
.# language. See https://github.com/imatix/gsl for details. This script
.# is licensed under MIT/X11.
.#
.output "./sock_option_test.go"
//go:generate gsl sockopts.xml
package goczmq
/* =========================================================================
zsock_option - get/set 0MQ socket options
****************************************************
* GENERATED SOURCE CODE, DO NOT EDIT!! *
****************************************************
Copyright (c) the Contributors as noted in the AUTHORS file.
This file is part of goczmq, the high-level go binding for CZMQ:
http://github.com/zeromq/goczmq
This Source Code Form is subject to the terms of the Mozilla Public
License, v. 2.0. If a copy of the MPL was not distributed with this
file, You can obtain one at http://mozilla.org/MPL/2.0/.
=========================================================================
*/
import (
"testing"
)
.for version
.if major = "4"
.for option where defined(test)
.if mode = "rw" | mode = "w"
.if type = "uint64" | type = "int64" | type = "uint32" | type = "int"
func Test$(name:pascal)(t *testing.T) {
sock := NewSock($(test:neat))
testval := $(test_value?'1':)
sock.Set$(name:pascal)(testval)
.if mode = "rw"
val := sock.$(name:pascal)()
if val != testval && val != 0 {
t.Errorf("$(name:pascal) returned %d, should be %d", val, testval)
}
.endif
sock.Destroy()
}
.endif
.if type = "string" | type = "key"
func Test$(name:pascal)(t *testing.T) {
sock := NewSock($(test:neat))
testval := "$(test_value?'test':)"
sock.Set$(name:pascal)(testval)
.if mode = "rw"
val := sock.$(name:pascal)()
if val != testval && val != "" {
t.Errorf("$(name:pascal) returned %s should be %s", val, testval)
}
.endif
sock.Destroy()
}
.endif
.if mode = "r"
.if type = "uint64" | type = "int64" | type = "uint32" | type = "int"
func Test$(name:pascal)(t *testing.T) {
sock := NewSock($(test:neat))
_ = sock.$(name:pascal)()
sock.Destroy()
}
.endif
.if type = "string" | type = "key"
func Test$(name:pascal)(t *testing.T) {
sock := NewSock($(test:pascal))
_ = sock.$(name:pascal)()
sock.Destroy()
}
.endif
.endif
.endif
.endfor
.endif
.endfor

View File

@ -1,32 +0,0 @@
.# This is a code generator built using the iMatix GSL code generation
.# language. See https://github.com/imatix/gsl for details. This script
.# is licensed under MIT/X11.
.#
.template 0
for version
# Expand any macros
for include
for options.macro where name = include.name
for . as child
copy child to version
endfor
endfor
endfor
# Preprocess options
for option
if type = "uint64" | type = "int64" | type = "uint32" | type = "int"
option.ctype = "int"
option.ctype_const = "int"
option.gotype = "int"
elsif type = "string" | type = "key"
option.ctype = "char *" # Enforce C strings
option.ctype_const = "const char *"
option.gotype = "string"
else
echo "E: unknown type: $(type)"
endif
endfor
endfor
.endtemplate
.include "sock_option.gsl"
.include "sock_option_test.gsl"

View File

@ -1,233 +0,0 @@
<?xml?>
<!-- Used to generate the socket options interface
Requires gsl4 from https://github.com/imatix/gsl
use 'gsl sockopts'
-->
<options script = "sockopts">
<version major = "4" minor = "2" style = "macro">
<!-- Options that are new in 4.2 -->
<option name = "heartbeat_ivl" type = "int" mode = "rw" test = "DEALER"
test_value = "2000" />
<option name = "heartbeat_ttl" type = "int" mode = "rw" test = "DEALER"
test_value = "4000" />
<option name = "heartbeat_timeout" type = "int" mode = "rw" test = "DEALER"
test_value = "6000" />
<option name = "use_fd" type = "int" mode = "rw" test = "REQ"
test_value = "3" />
<option name = "xpub_manual" type = "int" mode = "w" test = "XPUB"
test_value = "1" >
<restrict type = "XPUB" />
</option>
<option name = "xpub_welcome_msg" type = "string" mode = "w" test = "XPUB"
test_value = "welcome" >
<restrict type = "XPUB" />
</option>
<option name = "stream_notify" type = "int" mode = "w" test = "STREAM"
test_value = "1" >
<restrict type = "STREAM" />
</option>
<option name = "invert_matching" type = "int" mode = "rw" test = "XPUB"
test_value = "1" >
<restrict type = "XPUB" />
<restrict type = "PUB" />
<restrict type = "SUB" />
</option>
<option name = "xpub_verboser" type = "int" mode = "w" test = "XPUB"
test_value = "1" >
<restrict type = "XPUB" />
</option>
<option name = "connect_timeout" type = "int" mode = "rw" test = "DEALER"
test_value = "200" />
<option name = "tcp_maxrt" type = "int" mode = "rw" test = "DEALER"
test_value = "200" />
<option name = "thread_safe" type = "int" mode = "r" test = "DEALER"
test_value = "0" />
<option name = "multicast_maxtpdu" type = "int" mode = "rw" test = "DEALER"
test_value = "1400" />
<!-- We don't test these as libzmq doesn't always support VMCI -->
<option name = "vmci_buffer_size" type = "uint64" mode = "rw" />
<option name = "vmci_buffer_min_size" type = "uint64" mode = "rw" />
<option name = "vmci_buffer_max_size" type = "uint64" mode = "rw" />
<option name = "vmci_connect_timeout" type = "int" mode = "rw" />
</version>
<version major = "4" minor = "1" style = "macro">
<!-- Options that are new in 4.1 -->
<option name = "tos" type = "int" mode = "rw" test = "DEALER" />
<option name = "router_handover" type = "int" mode = "w" test = "ROUTER">
<restrict type = "ROUTER" />
</option>
<option name = "connect_rid" type = "key" mode = "w" test = "ROUTER"
test_value = "ABCD" >
<restrict type = "ROUTER" />
<restrict type = "STREAM" />
</option>
<option name = "handshake_ivl" type = "int" mode = "rw" test = "DEALER"
test_value = "200" />
<option name = "socks_proxy" type = "string" mode = "rw" test = "DEALER"
test_value = "127.0.0.1" />
<option name = "xpub_nodrop" type = "int" mode = "w" test = "XPUB"
test_value = "1" >
<restrict type = "XPUB" />
<restrict type = "PUB" />
</option>
</version>
<version major = "4" style = "macro">
<!-- Options that are new in 4.1 -->
<option name = "tos" type = "int" mode = "rw" test = "Dealer" />
<option name = "router_handover" type = "int" mode = "w" test = "Router">
<restrict type = "Router" />
</option>
<!-- Options that are new in 4.0 -->
<option name = "router_mandatory" type = "int" mode = "w" test = "Router">
<restrict type = "Router" />
</option>
<option name = "probe_router" type = "int" mode = "w" test = "Dealer">
<restrict type = "Router" />
<restrict type = "Dealer" />
<restrict type = "Req" />
</option>
<option name = "req_relaxed" type = "int" mode = "w" test = "Req">
<restrict type = "Req" />
</option>
<option name = "req_correlate" type = "int" mode = "w" test = "Req">
<restrict type = "Req" />
</option>
<option name = "conflate" type = "int" mode = "w" test = "Push">
<restrict type = "Push" />
<restrict type = "Pull" />
<restrict type = "Pub" />
<restrict type = "Sub" />
<restrict type = "Dealer" />
</option>
<!-- Security options -->
<option name = "zap_domain" type = "string" mode = "rw" test = "Sub" />
<option name = "mechanism" type = "int" mode = "r" test = "Sub" />
<option name = "plain_server" type = "int" mode = "rw" test = "Pub" />
<option name = "plain_username" type = "string" mode = "rw" test = "Sub" />
<option name = "plain_password" type = "string" mode = "rw" test = "Sub" />
<!-- We don't test these as libzmq doesn't always support CURVE security -->
<option name = "curve_server" type = "int" mode = "rw" />
<option name = "curve_publickey" type = "key" mode = "rw" />
<option name = "curve_secretkey" type = "key" mode = "rw" />
<option name = "curve_serverkey" type = "key" mode = "rw" />
<!-- We don't test these as libzmq doesn't always support GSSAPI security -->
<option name = "gssapi_server" type = "int" mode = "rw" />
<option name = "gssapi_plaintext" type = "int" mode = "rw" />
<option name = "gssapi_principal" type = "string" mode = "rw" />
<option name = "gssapi_service_principal"
type = "string" mode = "rw" />
<!-- New names for deprecated 3.x options -->
<option name = "ipv6" type = "int" mode = "rw" test = "Sub" />
<option name = "immediate" type = "int" mode = "rw" test = "Dealer" />
<!-- Deprecated 3.x options -->
<option name = "router_raw" type = "int" mode = "w" test = "Router">
<restrict type = "Router" />
</option>
<option name = "ipv4only" type = "int" mode = "rw" test = "Sub" />
<option name = "delay_attach_on_connect"
type = "int" mode = "w" test = "Pub" />
<!-- Options that are the same in 3.x -->
<include name = "3-x options" />
</version>
<version major = "3" style = "macro">
<!-- Options that are carried forward to 4.0 -->
<include name = "3-x options" />
<!-- Options that are deprecated in 4.0 -->
<option name = "router_raw" type = "int" mode = "w" test = "Router">
<restrict type = "Router" />
</option>
<option name = "ipv4only" type = "int" mode = "rw" test = "Sub" />
<option name = "delay_attach_on_connect"
type = "int" mode = "w" test = "Pub" />
</version>
<!-- Legacy version 2 -->
<version major = "2" style = "macro">
<option name = "hwm" type = "uint64" mode = "rw" test = "Sub" />
<option name = "swap" type = "int64" mode = "rw" test = "Sub" />
<option name = "affinity" type = "uint64" mode = "rw" test = "Sub" />
<option name = "identity" type = "string" mode = "rw" test = "Sub" />
<option name = "rate" type = "int64" mode = "rw" test = "Sub" />
<option name = "recovery_ivl" type = "int64" mode = "rw" test = "Sub" />
<option name = "recovery_ivl_msec" type = "int64" mode = "rw" test = "Sub" />
<option name = "mcast_loop" type = "int64" mode = "rw" test = "Sub" />
<option name = "rcvtimeo" type = "int" mode = "rw" test = "Sub" minor = "2" />
<option name = "sndtimeo" type = "int" mode = "rw" test = "Sub" minor = "2" />
<option name = "sndbuf" type = "uint64" mode = "rw" test = "Sub" />
<option name = "rcvbuf" type = "uint64" mode = "rw" test = "Sub" />
<option name = "linger" type = "int" mode = "rw" test = "Sub" />
<option name = "reconnect_ivl" type = "int" mode = "rw" test = "Sub" />
<option name = "reconnect_ivl_max" type = "int" mode = "rw" test = "Sub" />
<option name = "backlog" type = "int" mode = "rw" test = "Sub" />
<option name = "subscribe" type = "string" mode = "w" test = "Sub">
<restrict type = "Sub" />
</option>
<option name = "unsubscribe" type = "string" mode = "w" test = "Sub">
<restrict type = "Sub" />
</option>
<option name = "type" type = "int" mode = "r" test = "Sub" />
<option name = "rcvmore" type = "int64" mode = "r" test = "Sub" />
<option name = "fd" type = "int" mode = "r" test = "Sub" />
<option name = "events" type = "uint32" mode = "r" test = "Sub" />
</version>
<macro name = "3-x options">
<option name = "type" type = "int" mode = "r" test = "Sub" />
<option name = "sndhwm" type = "int" mode = "rw" test = "Pub" />
<option name = "rcvhwm" type = "int" mode = "rw" test = "Sub" />
<option name = "affinity" type = "uint64" mode = "rw" test = "Sub" />
<option name = "subscribe" type = "string" mode = "w" test = "Sub">
<restrict type = "Sub" />
</option>
<option name = "unsubscribe" type = "string" mode = "w" test = "Sub">
<restrict type = "Sub" />
</option>
<option name = "identity" type = "string" mode = "rw" test = "Dealer">
<restrict type = "Req" />
<restrict type = "Rep" />
<restrict type = "Dealer" />
<restrict type = "Router" />
</option>
<option name = "rate" type = "int" mode = "rw" test = "Sub" />
<option name = "recovery_ivl" type = "int" mode = "rw" test = "Sub" />
<option name = "sndbuf" type = "int" mode = "rw" test = "Pub" />
<option name = "rcvbuf" type = "int" mode = "rw" test = "Sub" />
<option name = "linger" type = "int" mode = "rw" test = "Sub" />
<option name = "reconnect_ivl" type = "int" mode = "rw" test = "Sub" />
<option name = "reconnect_ivl_max" type = "int" mode = "rw" test = "Sub" />
<option name = "backlog" type = "int" mode = "rw" test = "Sub" />
<option name = "maxmsgsize" type = "int64" mode = "rw" test = "Sub" />
<option name = "multicast_hops" type = "int" mode = "rw" test = "Sub" />
<option name = "rcvtimeo" type = "int" mode = "rw" test = "Sub" />
<option name = "sndtimeo" type = "int" mode = "rw" test = "Sub" />
<option name = "xpub_verbose" type = "int" mode = "w" test = "XPub">
<restrict type = "XPub" />
</option>
<option name = "tcp_keepalive" type = "int" mode = "rw" test = "Sub" />
<option name = "tcp_keepalive_idle"
type = "int" mode = "rw" test = "Sub" />
<option name = "tcp_keepalive_cnt" type = "int" mode = "rw" test = "Sub" />
<option name = "tcp_keepalive_intvl"
type = "int" mode = "rw" test = "Sub" />
<option name = "tcp_accept_filter" type = "string" mode = "rw" test = "Sub"
test_value = "127.0.0.1" />
<option name = "rcvmore" type = "int" mode = "r" test = "Sub" />
<option name = "fd" type = "int" mode = "r" test = "Sub" />
<option name = "events" type = "int" mode = "r" test = "Sub" />
<option name = "last_endpoint" type = "string" mode = "r" test = "Sub" />
</macro>
</options>

3
vendor/modules.txt vendored
View File

@ -89,9 +89,6 @@ golang.org/x/text/unicode/norm
# gopkg.in/ini.v1 v1.67.0
## explicit
gopkg.in/ini.v1
# gopkg.in/zeromq/goczmq.v4 v4.1.0
## explicit
gopkg.in/zeromq/goczmq.v4
# xorm.io/builder v0.3.12
## explicit; go 1.11
xorm.io/builder