Skip to content

Include some of czmq features needed for discovery #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Asmod4n merged 6 commits into from
Jun 11, 2017
Merged

Include some of czmq features needed for discovery #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
67a49d8
always copy data from a zmq message
Asmod4n Jun 11, 2017
675ef73
add ability to get broadcast or multicast address of a local interface
Asmod4n Jun 11, 2017
5966d18
move Timers to ZMQ Namespace
Asmod4n Jun 11, 2017
a77bb46
add support for getting broadcast addresses
Asmod4n Jun 11, 2017
8949ff4
copy zmq messages
Asmod4n Jun 11, 2017
660f3c9
remove unneeded flag
Asmod4n Jun 11, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion mrbgem.rake
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ MRuby::Gem::Specification.new('mruby-zmq') do |spec|
sh "cd #{spec.dir}/libzmq && ./autogen.sh && mkdir -p #{spec.build_dir}/build && cd #{spec.build_dir}/build && #{spec.dir}/libzmq/configure CC=#{spec.cc.command} CFLAGS=\"#{spec.cc.flags.join(' ')}\" LDFLAGS=\"#{spec.linker.flags.join(' ')}\" CXX=#{spec.cxx.command} CXXFLAGS=\"#{spec.cxx.flags.join(' ')}\" --host=#{build.host_target} --build=#{build.build_target} --disable-shared --enable-static --without-docs --prefix=#{spec.build_dir} && make -j4 && make install"
end
spec.linker.flags_before_libraries << "\"#{spec.build_dir}/lib/libzmq.a\" -pthread"
spec.linker.libraries << 'stdc++' << 'pthread'
spec.linker.libraries << 'stdc++'
spec.cc.include_paths << "#{spec.build_dir}/include"
spec.cxx.include_paths << "#{spec.build_dir}/include"
build.cc.include_paths << "#{spec.build_dir}/include"
Expand Down
6 changes: 3 additions & 3 deletions mrblib/thread.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ class Thread

def new(mrb_class, *args, &block)
LibZMQ.send(@pipe, [NEW, mrb_class, args, block].to_msgpack, 0)
msg = MessagePack.unpack(@pipe.recv.to_str(true))
msg = MessagePack.unpack(@pipe.recv.to_str)
case msg[TYPE]
when INSTANCE
ThreadProxy.new(self, msg[1])
Expand All @@ -77,7 +77,7 @@ def new(mrb_class, *args, &block)

def send(object_id, method, *args, &block)
LibZMQ.send(@pipe, [SEND, object_id, method, args, block].to_msgpack, 0)
msg = MessagePack.unpack(@pipe.recv.to_str(true))
msg = MessagePack.unpack(@pipe.recv.to_str)
case msg[TYPE]
when RESULT
msg[1]
Expand Down Expand Up @@ -198,7 +198,7 @@ def run
TERM = "TERM$".freeze

def handle_pipe
msg = @pipe.recv.to_str(true)
msg = @pipe.recv.to_str
if msg == TERM
@interrupted = true
else
Expand Down
135 changes: 119 additions & 16 deletions src/mrb_libzmq.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -302,17 +302,7 @@ mrb_zmq_proxy_steerable(mrb_state *mrb, mrb_value self)
static mrb_value
mrb_zmq_msg_to_str(mrb_state *mrb, mrb_value self)
{
mrb_bool static_string = FALSE;
mrb_get_args(mrb, "|b", &static_string);

void *data = zmq_msg_data((zmq_msg_t *) DATA_PTR(self));
size_t size = zmq_msg_size((zmq_msg_t *) DATA_PTR(self));

if (static_string) {
return mrb_str_new_static(mrb, (const char *) data, size);
} else {
return mrb_str_new(mrb, (const char *) data, size);
}
return mrb_str_new(mrb, (const char *) zmq_msg_data((zmq_msg_t *) DATA_PTR(self)), zmq_msg_size((zmq_msg_t *) DATA_PTR(self)));
}

static mrb_value
Expand Down Expand Up @@ -1110,11 +1100,12 @@ mrb_zmq_timers_execute(mrb_state *mrb, mrb_value self)

#ifdef HAVE_IFADDRS_H
MRB_INLINE mrb_bool
s_valid_flags (unsigned int flags)
s_valid_flags (unsigned int flags, int is_ipv6)
{
return (flags & IFF_UP) // Only use interfaces that are running
&& !(flags & IFF_LOOPBACK) // Ignore loopback interface
&& (flags & IFF_BROADCAST) // Only use interfaces that have BROADCAST
&& ((is_ipv6 || (flags & IFF_BROADCAST)) // Only use interfaces that have BROADCAST
&& (!is_ipv6 || (flags & IFF_MULTICAST))) // or IPv6 and MULTICAST
# if defined (IFF_SLAVE)
&& !(flags & IFF_SLAVE) // Ignore devices that are bonding slaves.
# endif
Expand All @@ -1141,7 +1132,7 @@ mrb_network_interfaces(mrb_state *mrb, mrb_value self)
if (likely(getifaddrs(&interfaces) == 0)) {
struct ifaddrs *interface;
for (interface = interfaces; interface; interface = interface->ifa_next) {
if (s_valid_flags(interface->ifa_flags)
if (s_valid_flags(interface->ifa_flags, is_ipv6)
&& interface->ifa_addr
&& interface->ifa_addr->sa_family == (is_ipv6 ? AF_INET6 : AF_INET)
&& (interface->ifa_broadaddr || (is_ipv6 && (interface->ifa_addr->sa_family == AF_INET6)))) {
Expand All @@ -1164,6 +1155,117 @@ mrb_network_interfaces(mrb_state *mrb, mrb_value self)

return interfaces_ary;
}

static mrb_value
s_get_ipv4_broadcast(mrb_state *mrb, const char *ip)
{
struct sockaddr_in sa;
int rc = inet_pton(AF_INET, ip, &(sa.sin_addr));
if (likely(rc == 1)) {
struct mrb_jmpbuf* prev_jmp = mrb->jmp;
struct mrb_jmpbuf c_jmp;
struct ifaddrs *interfaces = NULL;

MRB_TRY(&c_jmp)
{
mrb->jmp = &c_jmp;
if (likely(getifaddrs(&interfaces) == 0)) {
struct ifaddrs *interface;
for (interface = interfaces; interface; interface = interface->ifa_next) {
if (s_valid_flags(interface->ifa_flags, FALSE)
&& interface->ifa_addr
&& interface->ifa_addr->sa_family == AF_INET
&& interface->ifa_broadaddr
&& ((struct sockaddr_in *)interface->ifa_addr)->sin_addr.s_addr == sa.sin_addr.s_addr) {
mrb_value ipv4_str = mrb_str_new(mrb, NULL, INET_ADDRSTRLEN);
const char *rc = inet_ntop(AF_INET, &(((struct sockaddr_in *)interface->ifa_broadaddr)->sin_addr), RSTRING_PTR(ipv4_str), RSTRING_CAPA(ipv4_str));
if (likely(rc)) {
return mrb_str_resize(mrb, ipv4_str, strlen(rc));
} else {
mrb_sys_fail(mrb, "inet_ntop");
}
}
}
freeifaddrs(interfaces);
} else {
mrb_sys_fail(mrb, "getifaddrs");
}
mrb->jmp = prev_jmp;
}
MRB_CATCH(&c_jmp)
{
mrb->jmp = prev_jmp;
freeifaddrs(interfaces);
MRB_THROW(mrb->jmp);
}
MRB_END_EXC(&c_jmp);
} else if (rc == 0) {
mrb_raise(mrb, E_ARGUMENT_ERROR, "IP Address is invalid");
} else {
mrb_sys_fail(mrb, "inet_pton");
}

mrb_raise(mrb, E_ARGUMENT_ERROR, "Cannot find interface for given IP Address");
}

static mrb_value
s_get_ipv6_multicast(mrb_state *mrb, const char *ip)
{
struct sockaddr_in6 sa;
int rc = inet_pton(AF_INET6, ip, &(sa.sin6_addr));
if (likely(rc == 1)) {
struct mrb_jmpbuf* prev_jmp = mrb->jmp;
struct mrb_jmpbuf c_jmp;
struct ifaddrs *interfaces = NULL;

MRB_TRY(&c_jmp)
{
mrb->jmp = &c_jmp;
if (likely(getifaddrs(&interfaces) == 0)) {
struct ifaddrs *interface;
for (interface = interfaces; interface; interface = interface->ifa_next) {
if (s_valid_flags(interface->ifa_flags, TRUE)
&& interface->ifa_addr
&& interface->ifa_addr->sa_family == AF_INET6
&& memcmp(((struct sockaddr_in6 *)interface->ifa_addr)->sin6_addr.s6_addr, sa.sin6_addr.s6_addr, sizeof(sa.sin6_addr.s6_addr)) == 0) {
return mrb_str_cat_cstr(mrb, mrb_str_new_lit(mrb, "ff02::1%"), interface->ifa_name);
}
}
freeifaddrs(interfaces);
} else {
mrb_sys_fail(mrb, "getifaddrs");
}
mrb->jmp = prev_jmp;
}
MRB_CATCH(&c_jmp)
{
mrb->jmp = prev_jmp;
freeifaddrs(interfaces);
MRB_THROW(mrb->jmp);
}
MRB_END_EXC(&c_jmp);
} else if (rc == 0) {
mrb_raise(mrb, E_ARGUMENT_ERROR, "IP Address is invalid");
} else {
mrb_sys_fail(mrb, "inet_pton");
}

mrb_raise(mrb, E_ARGUMENT_ERROR, "Cannot find interface for given IP Address");
}

static mrb_value
mrb_zmq_get_broadcast_address(mrb_state *mrb, mrb_value self)
{
char *ip;
mrb_bool is_ipv6;
mrb_get_args(mrb, "zb", &ip, &is_ipv6);

if (is_ipv6)
return s_get_ipv6_multicast(mrb, ip);
else
return s_get_ipv4_broadcast(mrb, ip);
}

#endif //HAVE_IFADDRS_H

static mrb_value
Expand Down Expand Up @@ -1265,7 +1367,7 @@ mrb_mruby_zmq_gem_init(mrb_state* mrb)
zmq_msg_class = mrb_define_class_under(mrb, zmq_mod, "Msg", mrb->object_class);
MRB_SET_INSTANCE_TT(zmq_msg_class, MRB_TT_DATA);
mrb_define_method(mrb, zmq_msg_class, "initialize", mrb_zmq_msg_new, MRB_ARGS_NONE());
mrb_define_method(mrb, zmq_msg_class, "to_str", mrb_zmq_msg_to_str, MRB_ARGS_OPT(1));
mrb_define_method(mrb, zmq_msg_class, "to_str", mrb_zmq_msg_to_str, MRB_ARGS_NONE());
mrb_define_method(mrb, zmq_msg_class, "==", mrb_zmq_msg_eql, MRB_ARGS_REQ(1));

zmq_socket_class = mrb_define_class_under(mrb, zmq_mod, "Socket", mrb->object_class);
Expand All @@ -1290,7 +1392,7 @@ mrb_mruby_zmq_gem_init(mrb_state* mrb)

#ifdef ZMQ_HAVE_TIMERS
struct RClass *zmq_timers_class, *zmq_timers_timer_fn_class;
zmq_timers_class = mrb_define_class_under(mrb, libzmq_mod, "Timers", mrb->object_class);
zmq_timers_class = mrb_define_class_under(mrb, zmq_mod, "Timers", mrb->object_class);
MRB_SET_INSTANCE_TT(zmq_timers_class, MRB_TT_DATA);
mrb_define_method(mrb, zmq_timers_class, "initialize", mrb_zmq_timers_new, MRB_ARGS_NONE());
mrb_define_method(mrb, zmq_timers_class, "add", mrb_zmq_timers_add, (MRB_ARGS_REQ(1)|MRB_ARGS_BLOCK()));
Expand All @@ -1305,6 +1407,7 @@ mrb_mruby_zmq_gem_init(mrb_state* mrb)

#ifdef HAVE_IFADDRS_H
mrb_define_module_function(mrb, zmq_mod, "network_interfaces", mrb_network_interfaces, MRB_ARGS_NONE());
mrb_define_module_function(mrb, zmq_mod, "get_broadcast_address", mrb_zmq_get_broadcast_address, MRB_ARGS_REQ(2));
#endif

#define mrb_zmq_define_const(ZMQ_CONST_NAME, ZMQ_CONST) \
Expand Down
1 change: 1 addition & 0 deletions src/mrb_libzmq.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
#include <mruby/numeric.h>
#include <mruby/proc.h>
#ifdef HAVE_IFADDRS_H
#include <arpa/inet.h>
#include <net/if.h>
#include <ifaddrs.h>
#endif
Expand Down