Jeff Garzik
9 years ago
23 changed files with 850 additions and 3 deletions
@ -0,0 +1,37 @@
@@ -0,0 +1,37 @@
|
||||
#!/usr/bin/env python2 |
||||
|
||||
import array |
||||
import binascii |
||||
import zmq |
||||
|
||||
port = 28332 |
||||
|
||||
zmqContext = zmq.Context() |
||||
zmqSubSocket = zmqContext.socket(zmq.SUB) |
||||
zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "hashblock") |
||||
zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "hashtx") |
||||
zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "rawblock") |
||||
zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "rawtx") |
||||
zmqSubSocket.connect("tcp://127.0.0.1:%i" % port) |
||||
|
||||
try: |
||||
while True: |
||||
msg = zmqSubSocket.recv_multipart() |
||||
topic = str(msg[0]) |
||||
body = msg[1] |
||||
|
||||
if topic == "hashblock": |
||||
print "- HASH BLOCK -" |
||||
print binascii.hexlify(body) |
||||
elif topic == "hashtx": |
||||
print '- HASH TX -' |
||||
print binascii.hexlify(body) |
||||
elif topic == "rawblock": |
||||
print "- RAW BLOCK HEADER -" |
||||
print binascii.hexlify(body[:80]) |
||||
elif topic == "rawtx": |
||||
print '- RAW TX -' |
||||
print binascii.hexlify(body) |
||||
|
||||
except KeyboardInterrupt: |
||||
zmqContext.destroy() |
||||
@ -0,0 +1,26 @@
@@ -0,0 +1,26 @@
|
||||
package=zeromq |
||||
$(package)_version=4.0.4 |
||||
$(package)_download_path=http://download.zeromq.org |
||||
$(package)_file_name=$(package)-$($(package)_version).tar.gz |
||||
$(package)_sha256_hash=1ef71d46e94f33e27dd5a1661ed626cd39be4d2d6967792a275040e34457d399 |
||||
|
||||
define $(package)_set_vars |
||||
$(package)_config_opts=--without-documentation --disable-shared |
||||
$(package)_config_opts_linux=--with-pic |
||||
endef |
||||
|
||||
define $(package)_config_cmds |
||||
$($(package)_autoconf) |
||||
endef |
||||
|
||||
define $(package)_build_cmds |
||||
$(MAKE) -C src |
||||
endef |
||||
|
||||
define $(package)_stage_cmds |
||||
$(MAKE) -C src DESTDIR=$($(package)_staging_dir) install |
||||
endef |
||||
|
||||
define $(package)_postprocess_cmds |
||||
rm -rf bin share |
||||
endef |
||||
@ -0,0 +1,98 @@
@@ -0,0 +1,98 @@
|
||||
# Block and Transaction Broadcasting With ZeroMQ |
||||
|
||||
[ZeroMQ](http://zeromq.org/) is a lightweight wrapper around TCP |
||||
connections, inter-process communications, and shared-memory, |
||||
providing various message-oriented semantics such as publish/subcribe, |
||||
request/reply, and push/pull. |
||||
|
||||
The Bitcoin Core daemon can be configured to act as a trusted "border |
||||
router", implementing the bitcoin wire protocol and relay, making |
||||
consensus decisions, maintaining the local blockchain database, |
||||
broadcasting locally generated transactions into the network, and |
||||
providing a queryable RPC interface to interact on a polled basis for |
||||
requesting blockchain related data. However, there exists only a |
||||
limited service to notify external software of events like the arrival |
||||
of new blocks or transactions. |
||||
|
||||
The ZeroMQ facility implements a notification interface through a |
||||
set of specific notifiers. Currently there are notifiers that publish |
||||
blocks and transactions. This read-only facility requires only the |
||||
connection of a corresponding ZeroMQ subscriber port in receiving |
||||
software; it is not authenticated nor is there any two-way protocol |
||||
involvement. Therefore, subscribers should validate the received data |
||||
since it may be out of date, incomplete or even invalid. |
||||
|
||||
ZeroMQ sockets are self-connecting and self-healing; that is, connects |
||||
made between two endpoints will be automatically restored after an |
||||
outage, and either end may be freely started or stopped in any order. |
||||
|
||||
Because ZeroMQ is message oriented, subscribers receive transactions |
||||
and blocks all-at-once and do not need to implement any sort of |
||||
buffering or reassembly. |
||||
|
||||
## Prerequisites |
||||
|
||||
The ZeroMQ feature in Bitcoin Core uses only a very small part of the |
||||
ZeroMQ C API, and is thus compatible with any version of ZeroMQ |
||||
from 2.1 onward, including all versions in the 3.x and 4.x release |
||||
series. Typically, it is packaged by distributions as something like |
||||
*libzmq-dev*. |
||||
|
||||
The C++ wrapper for ZeroMQ is *not* needed. |
||||
|
||||
## Enabling |
||||
|
||||
By default, the ZeroMQ port functionality is enabled. Two steps are |
||||
required to enable--compiling in the ZeroMQ code, and configuring |
||||
runtime operation on the command-line or configuration file. |
||||
|
||||
$ ./configure --enable-zmq (other options) |
||||
|
||||
This will produce a binary that is capable of providing the ZeroMQ |
||||
facility, but will not do so until also configured properly. |
||||
|
||||
## Usage |
||||
|
||||
Currently, the following notifications are supported: |
||||
|
||||
-zmqpubhashtx=address |
||||
-zmqpubhashblock=address |
||||
-zmqpubrawblock=address |
||||
-zmqpubrawtx=address |
||||
|
||||
The socket type is PUB and the address must be a valid ZeroMQ |
||||
socket address. The same address can be used in more than one notification. |
||||
|
||||
For instance: |
||||
|
||||
$ bitcoind -zmqpubhashtx=tcp://127.0.0.1:28332 -zmqpubrawtx=ipc:///tmp/bitcoind.tx.raw |
||||
|
||||
Each PUB notification has a topic and body, where the header |
||||
corresponds to the notification type. For instance, for the notification |
||||
`-zmqpubhashtx` the topic is `hashtx` (no null terminator) and the body is the |
||||
hexadecimal transaction hash (32 bytes). |
||||
|
||||
These options can also be provided in bitcoin.conf. |
||||
|
||||
ZeroMQ endpoint specifiers for TCP (and others) are documented in the |
||||
[ZeroMQ API](http://api.zeromq.org). |
||||
|
||||
Client side, then, the ZeroMQ subscriber socket must have the |
||||
ZMQ_SUBSCRIBE option set to one or either of these prefixes (for instance, just `hash`); without |
||||
doing so will result in no messages arriving. Please see `contrib/zmq/zmq_sub.py` |
||||
for a working example. |
||||
|
||||
## Remarks |
||||
|
||||
From the perspective of bitcoind, the ZeroMQ socket is write-only; PUB |
||||
sockets don't even have a read function. Thus, there is no state |
||||
introduced into bitcoind directly. Furthermore, no information is |
||||
broadcast that wasn't already received from the public P2P network. |
||||
|
||||
No authentication or authorization is done on connecting clients; it |
||||
is assumed that the ZeroMQ port is exposed only to trusted entities, |
||||
using other means such as firewalling. |
||||
|
||||
Note that when the block chain tip changes, a reorganisation may occur and just |
||||
the tip will be notified. It is up to the subscriber to retrieve the chain |
||||
from the last known block to the new tip. |
||||
@ -0,0 +1,93 @@
@@ -0,0 +1,93 @@
|
||||
#!/usr/bin/env python2 |
||||
# Copyright (c) 2015 The Bitcoin Core developers |
||||
# Distributed under the MIT software license, see the accompanying |
||||
# file COPYING or http://www.opensource.org/licenses/mit-license.php. |
||||
|
||||
# |
||||
# Test ZMQ interface |
||||
# |
||||
|
||||
from test_framework.test_framework import BitcoinTestFramework |
||||
from test_framework.util import * |
||||
import zmq |
||||
import binascii |
||||
from test_framework.mininode import hash256 |
||||
|
||||
try: |
||||
import http.client as httplib |
||||
except ImportError: |
||||
import httplib |
||||
try: |
||||
import urllib.parse as urlparse |
||||
except ImportError: |
||||
import urlparse |
||||
|
||||
class ZMQTest (BitcoinTestFramework): |
||||
|
||||
port = 28332 |
||||
|
||||
def setup_nodes(self): |
||||
self.zmqContext = zmq.Context() |
||||
self.zmqSubSocket = self.zmqContext.socket(zmq.SUB) |
||||
self.zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "hashblock") |
||||
self.zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "hashtx") |
||||
self.zmqSubSocket.connect("tcp://127.0.0.1:%i" % self.port) |
||||
# Note: proxies are not used to connect to local nodes |
||||
# this is because the proxy to use is based on CService.GetNetwork(), which return NET_UNROUTABLE for localhost |
||||
return start_nodes(4, self.options.tmpdir, extra_args=[ |
||||
['-zmqpubhashtx=tcp://127.0.0.1:'+str(self.port), '-zmqpubhashblock=tcp://127.0.0.1:'+str(self.port)], |
||||
[], |
||||
[], |
||||
[] |
||||
]) |
||||
|
||||
def run_test(self): |
||||
self.sync_all() |
||||
|
||||
genhashes = self.nodes[0].generate(1); |
||||
self.sync_all() |
||||
|
||||
print "listen..." |
||||
msg = self.zmqSubSocket.recv_multipart() |
||||
topic = str(msg[0]) |
||||
body = msg[1] |
||||
|
||||
msg = self.zmqSubSocket.recv_multipart() |
||||
topic = str(msg[0]) |
||||
body = msg[1] |
||||
blkhash = binascii.hexlify(body) |
||||
|
||||
assert_equal(genhashes[0], blkhash) #blockhash from generate must be equal to the hash received over zmq |
||||
|
||||
n = 10 |
||||
genhashes = self.nodes[1].generate(n); |
||||
self.sync_all() |
||||
|
||||
zmqHashes = [] |
||||
for x in range(0,n*2): |
||||
msg = self.zmqSubSocket.recv_multipart() |
||||
topic = str(msg[0]) |
||||
body = msg[1] |
||||
if topic == "hashblock": |
||||
zmqHashes.append(binascii.hexlify(body)) |
||||
|
||||
for x in range(0,n): |
||||
assert_equal(genhashes[x], zmqHashes[x]) #blockhash from generate must be equal to the hash received over zmq |
||||
|
||||
#test tx from a second node |
||||
hashRPC = self.nodes[1].sendtoaddress(self.nodes[0].getnewaddress(), 1.0) |
||||
self.sync_all() |
||||
|
||||
#now we should receive a zmq msg because the tx was broadcastet |
||||
msg = self.zmqSubSocket.recv_multipart() |
||||
topic = str(msg[0]) |
||||
body = msg[1] |
||||
hashZMQ = "" |
||||
if topic == "hashtx": |
||||
hashZMQ = binascii.hexlify(body) |
||||
|
||||
assert_equal(hashRPC, hashZMQ) #blockhash from generate must be equal to the hash received over zmq |
||||
|
||||
|
||||
if __name__ == '__main__': |
||||
ZMQTest ().main () |
||||
@ -0,0 +1,22 @@
@@ -0,0 +1,22 @@
|
||||
// Copyright (c) 2015 The Bitcoin Core developers
|
||||
// Distributed under the MIT software license, see the accompanying
|
||||
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
|
||||
|
||||
#include "zmqabstractnotifier.h" |
||||
#include "util.h" |
||||
|
||||
|
||||
CZMQAbstractNotifier::~CZMQAbstractNotifier() |
||||
{ |
||||
assert(!psocket); |
||||
} |
||||
|
||||
bool CZMQAbstractNotifier::NotifyBlock(const uint256 &/*hash*/) |
||||
{ |
||||
return true; |
||||
} |
||||
|
||||
bool CZMQAbstractNotifier::NotifyTransaction(const CTransaction &/*transaction*/) |
||||
{ |
||||
return true; |
||||
} |
||||
@ -0,0 +1,42 @@
@@ -0,0 +1,42 @@
|
||||
// Copyright (c) 2015 The Bitcoin Core developers
|
||||
// Distributed under the MIT software license, see the accompanying
|
||||
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
|
||||
|
||||
#ifndef BITCOIN_ZMQ_ZMQABSTRACTNOTIFIER_H |
||||
#define BITCOIN_ZMQ_ZMQABSTRACTNOTIFIER_H |
||||
|
||||
#include "zmqconfig.h" |
||||
|
||||
class CZMQAbstractNotifier; |
||||
typedef CZMQAbstractNotifier* (*CZMQNotifierFactory)(); |
||||
|
||||
class CZMQAbstractNotifier |
||||
{ |
||||
public: |
||||
CZMQAbstractNotifier() : psocket(0) { } |
||||
virtual ~CZMQAbstractNotifier(); |
||||
|
||||
template <typename T> |
||||
static CZMQAbstractNotifier* Create() |
||||
{ |
||||
return new T(); |
||||
} |
||||
|
||||
std::string GetType() const { return type; } |
||||
void SetType(const std::string &t) { type = t; } |
||||
std::string GetAddress() const { return address; } |
||||
void SetAddress(const std::string &a) { address = a; } |
||||
|
||||
virtual bool Initialize(void *pcontext) = 0; |
||||
virtual void Shutdown() = 0; |
||||
|
||||
virtual bool NotifyBlock(const uint256 &hash); |
||||
virtual bool NotifyTransaction(const CTransaction &transaction); |
||||
|
||||
protected: |
||||
void *psocket; |
||||
std::string type; |
||||
std::string address; |
||||
}; |
||||
|
||||
#endif // BITCOIN_ZMQ_ZMQABSTRACTNOTIFIER_H
|
||||
@ -0,0 +1,24 @@
@@ -0,0 +1,24 @@
|
||||
// Copyright (c) 2015 The Bitcoin Core developers
|
||||
// Distributed under the MIT software license, see the accompanying
|
||||
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
|
||||
|
||||
#ifndef BITCOIN_ZMQ_ZMQCONFIG_H |
||||
#define BITCOIN_ZMQ_ZMQCONFIG_H |
||||
|
||||
#if defined(HAVE_CONFIG_H) |
||||
#include "config/bitcoin-config.h" |
||||
#endif |
||||
|
||||
#include <stdarg.h> |
||||
#include <string> |
||||
|
||||
#if ENABLE_ZMQ |
||||
#include <zmq.h> |
||||
#endif |
||||
|
||||
#include "primitives/block.h" |
||||
#include "primitives/transaction.h" |
||||
|
||||
void zmqError(const char *str); |
||||
|
||||
#endif // BITCOIN_ZMQ_ZMQCONFIG_H
|
||||
@ -0,0 +1,155 @@
@@ -0,0 +1,155 @@
|
||||
// Copyright (c) 2015 The Bitcoin Core developers
|
||||
// Distributed under the MIT software license, see the accompanying
|
||||
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
|
||||
|
||||
#include "zmqnotificationinterface.h" |
||||
#include "zmqpublishnotifier.h" |
||||
|
||||
#include "version.h" |
||||
#include "main.h" |
||||
#include "streams.h" |
||||
#include "util.h" |
||||
|
||||
void zmqError(const char *str) |
||||
{ |
||||
LogPrint("zmq", "Error: %s, errno=%s\n", str, zmq_strerror(errno)); |
||||
} |
||||
|
||||
CZMQNotificationInterface::CZMQNotificationInterface() : pcontext(NULL) |
||||
{ |
||||
} |
||||
|
||||
CZMQNotificationInterface::~CZMQNotificationInterface() |
||||
{ |
||||
// ensure Shutdown if Initialize is called
|
||||
assert(!pcontext); |
||||
|
||||
for (std::list<CZMQAbstractNotifier*>::iterator i=notifiers.begin(); i!=notifiers.end(); ++i) |
||||
{ |
||||
delete *i; |
||||
} |
||||
} |
||||
|
||||
CZMQNotificationInterface* CZMQNotificationInterface::CreateWithArguments(const std::map<std::string, std::string> &args) |
||||
{ |
||||
CZMQNotificationInterface* notificationInterface = NULL; |
||||
std::map<std::string, CZMQNotifierFactory> factories; |
||||
std::list<CZMQAbstractNotifier*> notifiers; |
||||
|
||||
factories["pubhashblock"] = CZMQAbstractNotifier::Create<CZMQPublishHashBlockNotifier>; |
||||
factories["pubhashtx"] = CZMQAbstractNotifier::Create<CZMQPublishHashTransactionNotifier>; |
||||
factories["pubrawblock"] = CZMQAbstractNotifier::Create<CZMQPublishRawBlockNotifier>; |
||||
factories["pubrawtx"] = CZMQAbstractNotifier::Create<CZMQPublishRawTransactionNotifier>; |
||||
|
||||
for (std::map<std::string, CZMQNotifierFactory>::const_iterator i=factories.begin(); i!=factories.end(); ++i) |
||||
{ |
||||
std::map<std::string, std::string>::const_iterator j = args.find("-zmq" + i->first); |
||||
if (j!=args.end()) |
||||
{ |
||||
CZMQNotifierFactory factory = i->second; |
||||
std::string address = j->second; |
||||
CZMQAbstractNotifier *notifier = factory(); |
||||
notifier->SetType(i->first); |
||||
notifier->SetAddress(address); |
||||
notifiers.push_back(notifier); |
||||
} |
||||
} |
||||
|
||||
if (!notifiers.empty()) |
||||
{ |
||||
notificationInterface = new CZMQNotificationInterface(); |
||||
notificationInterface->notifiers = notifiers; |
||||
} |
||||
|
||||
return notificationInterface; |
||||
} |
||||
|
||||
// Called at startup to conditionally set up ZMQ socket(s)
|
||||
bool CZMQNotificationInterface::Initialize() |
||||
{ |
||||
LogPrint("zmq", "Initialize notification interface\n"); |
||||
assert(!pcontext); |
||||
|
||||
pcontext = zmq_init(1); |
||||
|
||||
if (!pcontext) |
||||
{ |
||||
zmqError("Unable to initialize context"); |
||||
return false; |
||||
} |
||||
|
||||
std::list<CZMQAbstractNotifier*>::iterator i=notifiers.begin(); |
||||
for (; i!=notifiers.end(); ++i) |
||||
{ |
||||
CZMQAbstractNotifier *notifier = *i; |
||||
if (notifier->Initialize(pcontext)) |
||||
{ |
||||
LogPrint("zmq", " Notifier %s ready (address = %s)\n", notifier->GetType(), notifier->GetAddress()); |
||||
} |
||||
else |
||||
{ |
||||
LogPrint("zmq", " Notifier %s failed (address = %s)\n", notifier->GetType(), notifier->GetAddress()); |
||||
break; |
||||
} |
||||
} |
||||
|
||||
if (i!=notifiers.end()) |
||||
{ |
||||
Shutdown(); |
||||
return false; |
||||
} |
||||
|
||||
return false; |
||||
} |
||||
|
||||
// Called during shutdown sequence
|
||||
void CZMQNotificationInterface::Shutdown() |
||||
{ |
||||
LogPrint("zmq", "Shutdown notification interface\n"); |
||||
if (pcontext) |
||||
{ |
||||
for (std::list<CZMQAbstractNotifier*>::iterator i=notifiers.begin(); i!=notifiers.end(); ++i) |
||||
{ |
||||
CZMQAbstractNotifier *notifier = *i; |
||||
LogPrint("zmq", " Shutdown notifier %s at %s\n", notifier->GetType(), notifier->GetAddress()); |
||||
notifier->Shutdown(); |
||||
} |
||||
zmq_ctx_destroy(pcontext); |
||||
|
||||
pcontext = 0; |
||||
} |
||||
} |
||||
|
||||
void CZMQNotificationInterface::UpdatedBlockTip(const uint256 &hash) |
||||
{ |
||||
for (std::list<CZMQAbstractNotifier*>::iterator i = notifiers.begin(); i!=notifiers.end(); ) |
||||
{ |
||||
CZMQAbstractNotifier *notifier = *i; |
||||
if (notifier->NotifyBlock(hash)) |
||||
{ |
||||
i++; |
||||
} |
||||
else |
||||
{ |
||||
notifier->Shutdown(); |
||||
i = notifiers.erase(i); |
||||
} |
||||
} |
||||
} |
||||
|
||||
void CZMQNotificationInterface::SyncTransaction(const CTransaction &tx, const CBlock *pblock) |
||||
{ |
||||
for (std::list<CZMQAbstractNotifier*>::iterator i = notifiers.begin(); i!=notifiers.end(); ) |
||||
{ |
||||
CZMQAbstractNotifier *notifier = *i; |
||||
if (notifier->NotifyTransaction(tx)) |
||||
{ |
||||
i++; |
||||
} |
||||
else |
||||
{ |
||||
notifier->Shutdown(); |
||||
i = notifiers.erase(i); |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,35 @@
@@ -0,0 +1,35 @@
|
||||
// Copyright (c) 2015 The Bitcoin Core developers
|
||||
// Distributed under the MIT software license, see the accompanying
|
||||
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
|
||||
|
||||
#ifndef BITCOIN_ZMQ_ZMQNOTIFICATIONINTERFACE_H |
||||
#define BITCOIN_ZMQ_ZMQNOTIFICATIONINTERFACE_H |
||||
|
||||
#include "validationinterface.h" |
||||
#include <string> |
||||
#include <map> |
||||
|
||||
class CZMQAbstractNotifier; |
||||
|
||||
class CZMQNotificationInterface : public CValidationInterface |
||||
{ |
||||
public: |
||||
virtual ~CZMQNotificationInterface(); |
||||
|
||||
static CZMQNotificationInterface* CreateWithArguments(const std::map<std::string, std::string> &args); |
||||
|
||||
bool Initialize(); |
||||
void Shutdown(); |
||||
|
||||
protected: // CValidationInterface
|
||||
void SyncTransaction(const CTransaction &tx, const CBlock *pblock); |
||||
void UpdatedBlockTip(const uint256 &newHashTip); |
||||
|
||||
private: |
||||
CZMQNotificationInterface(); |
||||
|
||||
void *pcontext; |
||||
std::list<CZMQAbstractNotifier*> notifiers; |
||||
}; |
||||
|
||||
#endif // BITCOIN_ZMQ_ZMQNOTIFICATIONINTERFACE_H
|
||||
@ -0,0 +1,172 @@
@@ -0,0 +1,172 @@
|
||||
// Copyright (c) 2015 The Bitcoin Core developers
|
||||
// Distributed under the MIT software license, see the accompanying
|
||||
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
|
||||
|
||||
#include "zmqpublishnotifier.h" |
||||
#include "main.h" |
||||
#include "util.h" |
||||
|
||||
static std::multimap<std::string, CZMQAbstractPublishNotifier*> mapPublishNotifiers; |
||||
|
||||
// Internal function to send multipart message
|
||||
static int zmq_send_multipart(void *sock, const void* data, size_t size, ...) |
||||
{ |
||||
va_list args; |
||||
va_start(args, size); |
||||
|
||||
while (1) |
||||
{ |
||||
zmq_msg_t msg; |
||||
|
||||
int rc = zmq_msg_init_size(&msg, size); |
||||
if (rc != 0) |
||||
{ |
||||
zmqError("Unable to initialize ZMQ msg"); |
||||
return -1; |
||||
} |
||||
|
||||
void *buf = zmq_msg_data(&msg); |
||||
memcpy(buf, data, size); |
||||
|
||||
data = va_arg(args, const void*); |
||||
|
||||
rc = zmq_msg_send(&msg, sock, data ? ZMQ_SNDMORE : 0); |
||||
if (rc == -1) |
||||
{ |
||||
zmqError("Unable to send ZMQ msg"); |
||||
zmq_msg_close(&msg); |
||||
return -1; |
||||
} |
||||
|
||||
zmq_msg_close(&msg); |
||||
|
||||
if (!data) |
||||
break; |
||||
|
||||
size = va_arg(args, size_t); |
||||
} |
||||
return 0; |
||||
} |
||||
|
||||
bool CZMQAbstractPublishNotifier::Initialize(void *pcontext) |
||||
{ |
||||
assert(!psocket); |
||||
|
||||
// check if address is being used by other publish notifier
|
||||
std::multimap<std::string, CZMQAbstractPublishNotifier*>::iterator i = mapPublishNotifiers.find(address); |
||||
|
||||
if (i==mapPublishNotifiers.end()) |
||||
{ |
||||
psocket = zmq_socket(pcontext, ZMQ_PUB); |
||||
if (!psocket) |
||||
{ |
||||
zmqError("Failed to create socket"); |
||||
return false; |
||||
} |
||||
|
||||
int rc = zmq_bind(psocket, address.c_str()); |
||||
if (rc!=0) |
||||
{ |
||||
zmqError("Failed to bind address"); |
||||
return false; |
||||
} |
||||
|
||||
// register this notifier for the address, so it can be reused for other publish notifier
|
||||
mapPublishNotifiers.insert(std::make_pair(address, this)); |
||||
return true; |
||||
} |
||||
else |
||||
{ |
||||
LogPrint("zmq", " Reuse socket for address %s\n", address); |
||||
|
||||
psocket = i->second->psocket; |
||||
mapPublishNotifiers.insert(std::make_pair(address, this)); |
||||
|
||||
return true; |
||||
} |
||||
} |
||||
|
||||
void CZMQAbstractPublishNotifier::Shutdown() |
||||
{ |
||||
assert(psocket); |
||||
|
||||
int count = mapPublishNotifiers.count(address); |
||||
|
||||
// remove this notifier from the list of publishers using this address
|
||||
typedef std::multimap<std::string, CZMQAbstractPublishNotifier*>::iterator iterator; |
||||
std::pair<iterator, iterator> iterpair = mapPublishNotifiers.equal_range(address); |
||||
|
||||
for (iterator it = iterpair.first; it != iterpair.second; ++it) |
||||
{ |
||||
if (it->second==this) |
||||
{ |
||||
mapPublishNotifiers.erase(it); |
||||
break; |
||||
} |
||||
} |
||||
|
||||
if (count == 1) |
||||
{ |
||||
LogPrint("zmq", "Close socket at address %s\n", address); |
||||
int linger = 0; |
||||
zmq_setsockopt(psocket, ZMQ_LINGER, &linger, sizeof(linger)); |
||||
zmq_close(psocket); |
||||
} |
||||
|
||||
psocket = 0; |
||||
} |
||||
|
||||
bool CZMQPublishHashBlockNotifier::NotifyBlock(const uint256 &hash) |
||||
{ |
||||
LogPrint("zmq", "Publish hash block %s\n", hash.GetHex()); |
||||
char data[32]; |
||||
for (unsigned int i = 0; i < 32; i++) |
||||
data[31 - i] = hash.begin()[i]; |
||||
int rc = zmq_send_multipart(psocket, "hashblock", 9, data, 32, 0); |
||||
return rc == 0; |
||||
} |
||||
|
||||
bool CZMQPublishHashTransactionNotifier::NotifyTransaction(const CTransaction &transaction) |
||||
{ |
||||
uint256 hash = transaction.GetHash(); |
||||
LogPrint("zmq", "Publish hash transaction %s\n", hash.GetHex()); |
||||
char data[32]; |
||||
for (unsigned int i = 0; i < 32; i++) |
||||
data[31 - i] = hash.begin()[i]; |
||||
int rc = zmq_send_multipart(psocket, "hashtx", 6, data, 32, 0); |
||||
return rc == 0; |
||||
} |
||||
|
||||
bool CZMQPublishRawBlockNotifier::NotifyBlock(const uint256 &hash) |
||||
{ |
||||
LogPrint("zmq", "Publish raw block %s\n", hash.GetHex()); |
||||
|
||||
CDataStream ss(SER_NETWORK, PROTOCOL_VERSION); |
||||
{ |
||||
LOCK(cs_main); |
||||
|
||||
CBlock block; |
||||
CBlockIndex* pblockindex = mapBlockIndex[hash]; |
||||
|
||||
if(!ReadBlockFromDisk(block, pblockindex)) |
||||
{ |
||||
zmqError("Can't read block from disk"); |
||||
return false; |
||||
} |
||||
|
||||
ss << block; |
||||
} |
||||
|
||||
int rc = zmq_send_multipart(psocket, "rawblock", 8, &(*ss.begin()), ss.size(), 0); |
||||
return rc == 0; |
||||
} |
||||
|
||||
bool CZMQPublishRawTransactionNotifier::NotifyTransaction(const CTransaction &transaction) |
||||
{ |
||||
uint256 hash = transaction.GetHash(); |
||||
LogPrint("zmq", "Publish raw transaction %s\n", hash.GetHex()); |
||||
CDataStream ss(SER_NETWORK, PROTOCOL_VERSION); |
||||
ss << transaction; |
||||
int rc = zmq_send_multipart(psocket, "rawtx", 5, &(*ss.begin()), ss.size(), 0); |
||||
return rc == 0; |
||||
} |
||||
@ -0,0 +1,41 @@
@@ -0,0 +1,41 @@
|
||||
// Copyright (c) 2015 The Bitcoin Core developers
|
||||
// Distributed under the MIT software license, see the accompanying
|
||||
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
|
||||
|
||||
#ifndef BITCOIN_ZMQ_ZMQPUBLISHNOTIFIER_H |
||||
#define BITCOIN_ZMQ_ZMQPUBLISHNOTIFIER_H |
||||
|
||||
#include "zmqabstractnotifier.h" |
||||
|
||||
class CZMQAbstractPublishNotifier : public CZMQAbstractNotifier |
||||
{ |
||||
public: |
||||
bool Initialize(void *pcontext); |
||||
void Shutdown(); |
||||
}; |
||||
|
||||
class CZMQPublishHashBlockNotifier : public CZMQAbstractPublishNotifier |
||||
{ |
||||
public: |
||||
bool NotifyBlock(const uint256 &hash); |
||||
}; |
||||
|
||||
class CZMQPublishHashTransactionNotifier : public CZMQAbstractPublishNotifier |
||||
{ |
||||
public: |
||||
bool NotifyTransaction(const CTransaction &transaction); |
||||
}; |
||||
|
||||
class CZMQPublishRawBlockNotifier : public CZMQAbstractPublishNotifier |
||||
{ |
||||
public: |
||||
bool NotifyBlock(const uint256 &hash); |
||||
}; |
||||
|
||||
class CZMQPublishRawTransactionNotifier : public CZMQAbstractPublishNotifier |
||||
{ |
||||
public: |
||||
bool NotifyTransaction(const CTransaction &transaction); |
||||
}; |
||||
|
||||
#endif // BITCOIN_ZMQ_ZMQPUBLISHNOTIFIER_H
|
||||
Loading…
Reference in new issue