rtyler
/
zeromq-Ada
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Working towards 3.2

This commit is contained in:
Per Sandberg 2013-01-07 11:48:35 +01:00
parent 09b3f06830
commit 06e9084ef3
25 changed files with 1505 additions and 984 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
examples/hwserver.adb
View File

@ -38,13 +38,12 @@ procedure HWServer is
inbuffer : Ada.Strings.Unbounded.Unbounded_String;
begin
-- Prepare our context and socket
Context.Initialize (1);
Socket.Initialize (Context, ZMQ.Sockets.REP);
Socket.Bind ("tcp://*:5555");
loop
-- Wait for next request from client
inbuffer := Socket.recv;
inbuffer := Socket.Recv;
Put_Line ("Received request:" & To_String (inbuffer));
-- Do some 'work'

7
examples/zmq-examples-client.adb
View File

@ -31,9 +31,6 @@ procedure ZMQ.examples.Client is
ctx : ZMQ.Contexts.Context;
s : ZMQ.Sockets.Socket;
begin
-- Initialise 0MQ context, requesting a single application thread
-- and a single I/O thread
ctx.Initialize (1);
-- Create a ZMQ_REP socket to receive requests and send replies
s.Initialize (ctx, Sockets.REQ);
@ -54,8 +51,8 @@ begin
resultset : ZMQ.Messages.Message;
begin
resultset.Initialize;
s.recv (resultset);
Put_Line ('"' & resultset.getData & '"');
s.Recv (resultset);
Put_Line ('"' & resultset.GetData & '"');
resultset.Finalize;
end;
end loop;

6
examples/zmq-examples-display.adb
View File

@ -32,15 +32,13 @@ procedure ZMQ.examples.Display is
begin
Context.Initialize (1);
Socket.Initialize (Context, Sockets.SUB);
Socket.Establish_message_filter ("");
Socket.Establish_Message_Filter ("");
Socket.Bind ("tcp://lo:5555");
Ada.Text_IO.Put_Line ("Connected");
Read_Loop : loop
declare
Buffer : constant String := Socket.recv;
Buffer : constant String := Socket.Recv;
begin
Ada.Text_IO.Put_Line (Buffer);
exit Read_Loop when Buffer = END_MESSAGE;

14
examples/zmq-examples-multi_thread_server.adb
View File

@ -24,8 +24,8 @@
with ZMQ.Sockets;
with ZMQ.Contexts;
with ZMQ.devices;
with Ada.Strings.Unbounded; use Ada.Strings.Unbounded;
with ZMQ.Proxys;
procedure ZMQ.examples.Multi_Thread_Server is
task type server_task (ctx : not null access ZMQ.Contexts.Context;
@ -39,7 +39,7 @@ procedure ZMQ.examples.Multi_Thread_Server is
s.Initialize (ctx.all, Sockets.REP);
s.Connect ("inproc://workers");
loop
msg := s.recv;
msg := s.Recv;
Append (msg, "<Served by thread:" & id'Img & ">");
s.Send (msg);
end loop;
@ -53,25 +53,19 @@ procedure ZMQ.examples.Multi_Thread_Server is
workers : ZMQ.Sockets.Socket;
clients : ZMQ.Sockets.Socket;
dev : ZMQ.devices.device;
begin
-- Initialise 0MQ context, requesting a single application thread
-- and a single I/O thread
ctx.Initialize (servers'Length + 1);
-- Create a ZMQ_REP socket to receive requests and send replies
workers.Initialize (ctx, Sockets.XREQ);
workers.Bind ("inproc://workers");
-- Bind to the TCP transport and port 5555 on the 'lo' interface
clients.Initialize (ctx, Sockets.XREP);
clients.Bind ("tcp://lo:5555");
workers.Bind ("tcp://lo:5555");
for i in servers'Range loop
servers (i) := new server_task (ctx'Access, i);
end loop;
dev.initialize (devices.Queue, clients, workers);
ZMQ.Proxys.Proxy (workers, workers);
end ZMQ.Examples.Multi_Thread_Server;

1
examples/zmq-examples-prompt.adb
View File

@ -32,7 +32,6 @@ procedure ZMQ.examples.prompt is
s : Sockets.Socket;
begin
ctx.Initialize (1);
s.Initialize (ctx, Sockets.PUB);
s.Connect ("tcp://localhost:5555");

10
examples/zmq-examples-server.adb
View File

@ -32,10 +32,6 @@ procedure ZMQ.examples.Server is
s : ZMQ.Sockets.Socket;
resultset_string : constant String := "OK";
begin
-- Initialise 0MQ context, requesting a single application thread
-- and a single I/O thread
ctx.Initialize (1);
-- Create a ZMQ_REP socket to receive requests and send replies
s.Initialize (ctx, Sockets.REP);
@ -48,14 +44,14 @@ begin
begin
query.Initialize;
-- Receive a message, blocks until one is available
s.recv (query);
s.Recv (query);
-- Process the query
Put_Line (query.getData);
Put_Line (query.GetData);
declare
-- Allocate a response message and fill in an example response
resultset : ZMQ.Messages.Message;
begin
resultset.Initialize (query.getData & "->" & resultset_string);
resultset.Initialize (query.GetData & "->" & resultset_string);
-- Send back our canned response
s.Send (resultset);
resultset.Finalize;

6
renames.py
View File

@ -29,6 +29,12 @@ def rename(p):
buffer=buffer.split("\n")
ret=[]
ret.append('--------------------------------------------------------------------')
ret.append('-- --')
ret.append('-- Do not edit, this file is automaticly generated from "zmq.h" --')
ret.append('-- --')
ret.append('--------------------------------------------------------------------')
ret.append('')
for line in buffer:
ret.append(fix(line))
if re.match("^package.*is",line):

16
src/libzmq.gpr Normal file
View File

@ -0,0 +1,16 @@
project libZMQ is
for Source_Files use ("zmq.h","zmq_utils.h");
for Languages use ("C");
ZMQ_PREFIX := external("ZMQ_PREFIX", project'Project_Dir & "../../");
for Source_Dirs use (ZMQ_PREFIX &"include");
for Externally_Built use "True";
package Linker is
for Linker_Options use ("-L" & ZMQ_PREFIX & "lib",
"-lzmq",
"-luuid",
"-lrt",
"-lpthread");
end Linker;
end libZMQ;

2
src/zmq-contexts.ads
View File

@ -36,7 +36,7 @@ package ZMQ.Contexts is
IO_THREADS_DFLT : constant := 1;
MAX_SOCKETS_DFLT : constant := 1024;
type Context is new Ada.Finalization.Limited_Controlled with private;
type Context is tagged limited private;
type Any_Context is access all Context'Class;

2
src/zmq-devices.adb
View File

@ -51,7 +51,7 @@ package body ZMQ.Devices is
is
begin
This.Impl :=
Low_Level.Zmq_Device (Map (Kind),
Low_Level.zmq_device (Map (Kind),
In_Socket.Get_Impl,
Out_Ocket.Get_Impl);
end Initialize;

2
src/zmq-devices.ads
View File

@ -34,7 +34,7 @@
private with Interfaces.C;
with ZMQ.Sockets;
package ZMQ.Devices is
pragma Obsolescent ("Use ZMQ.Proxys");
pragma Obsolescent ("Use ZMQ.Proxys;");
-- Devices are building blocks intended to serve as intermediate nodes
-- in complex messaging topologies.

6
src/zmq-low_level.ads
View File

@ -1,3 +1,9 @@
--------------------------------------------------------------------
-- --
-- Do not edit, this file is automaticly generated from "zmq.h" --
-- --
--------------------------------------------------------------------
pragma Ada_2005;
pragma Style_Checks (Off);

12
src/zmq-messages.adb
View File

@ -45,7 +45,7 @@ package body ZMQ.Messages is
procedure Initialize (Self : in out Message) is
Ret : int;
begin
Ret := Low_Level.Zmq_Msg_Init (Self.Msg'Access);
Ret := Low_Level.zmq_msg_init (Self.Msg'Access);
if Ret /= 0 then
raise ZMQ_Error with Error_Message (GNAT.OS_Lib.Errno) & " in " &
GNAT.Source_Info.Enclosing_Entity;
@ -59,7 +59,7 @@ package body ZMQ.Messages is
procedure Initialize (Self : in out Message; Size : Natural) is
Ret : int;
begin
Ret := Low_Level.Zmq_Msg_Init_Size (Self.Msg'Access, size_t (Size));
Ret := Low_Level.zmq_msg_init_size (Self.Msg'Access, size_t (Size));
if Ret /= 0 then
raise ZMQ_Error with Error_Message (GNAT.OS_Lib.Errno) & " in " &
GNAT.Source_Info.Enclosing_Entity;
@ -111,7 +111,7 @@ package body ZMQ.Messages is
is
Ret : int;
begin
Ret := Low_Level.Zmq_Msg_Init_Data (Self.Msg'Access,
Ret := Low_Level.zmq_msg_init_data (Self.Msg'Access,
Message,
size_t (Size),
Free,
@ -187,7 +187,7 @@ package body ZMQ.Messages is
---------------
function GetData (Self : Message) return System.Address is
begin
return Low_Level.Zmq_Msg_Data (Self.Msg'Unrestricted_Access);
return Low_Level.zmq_msg_data (Self.Msg'Unrestricted_Access);
end GetData;
@ -242,7 +242,7 @@ package body ZMQ.Messages is
---------------
function GetSize (Self : Message) return Natural is
begin
return Natural (Low_Level.Zmq_Msg_Size (Self.Msg'Unrestricted_Access));
return Natural (Low_Level.zmq_msg_size (Self.Msg'Unrestricted_Access));
end GetSize;
@ -253,7 +253,7 @@ package body ZMQ.Messages is
procedure Finalize (Self : in out Message) is
Ret : int;
begin
Ret := Low_Level.Zmq_Msg_Close (Self.Msg'Access);
Ret := Low_Level.zmq_msg_close (Self.Msg'Access);
if Ret /= 0 then
raise ZMQ_Error with Error_Message (GNAT.OS_Lib.Errno) & " in " &
GNAT.Source_Info.Enclosing_Entity;

4
src/zmq-messages.ads
View File

@ -111,7 +111,7 @@ package ZMQ.Messages is
procedure Finalize (Self : in out Message);
type Zmq_Msg_T_Access is access all ZMQ.Low_Level.Zmq_Msg_T;
type Zmq_Msg_T_Access is access all ZMQ.Low_Level.zmq_msg_t;
function GetImpl (Self : Message) return not null Zmq_Msg_T_Access;
@ -139,6 +139,6 @@ package ZMQ.Messages is
private
type Message is tagged limited record
Msg : aliased ZMQ.Low_Level.Zmq_Msg_T;
Msg : aliased ZMQ.Low_Level.zmq_msg_t;
end record;
end ZMQ.Messages;

8
src/zmq-sockets-streams.ads
View File

@ -3,9 +3,7 @@ with Ada.Streams.Stream_IO; use Ada.Streams.Stream_IO;
package ZMQ.Sockets.Streams is
type Stream_Access is access all Root_Stream_Type'Class;
type Stream_Socket (With_Context : Contexts.Any_Context;
Kind : Socket_Type;
Buffer_Size : Positive) is new Socket with private;
type Stream_Socket (Buffer_Size : Positive) is new Socket with private;
function stream (this : Stream_Socket) return Stream_Access;
@ -32,9 +30,7 @@ package ZMQ.Sockets.Streams is
private
type Stream_Socket
(With_Context : Contexts.Any_Context;
Kind : Socket_Type;
Buffer_Size : Positive) is new Socket (With_Context, Kind) with
(Buffer_Size : Positive) is new Socket with
null record;
end ZMQ.Sockets.Streams;

733
src/zmq-sockets.adb
View File

@ -33,6 +33,7 @@ with ZMQ.Low_Level;
with Interfaces.C.Strings;
with GNAT.Source_Info;
with Ada.Strings.Unbounded; use Ada.Strings.Unbounded;
with System.Address_To_Access_Conversions;
package body ZMQ.Sockets is
use Interfaces.C.Strings;
use Interfaces.C;
@ -40,70 +41,55 @@ package body ZMQ.Sockets is
use Ada.Streams;
type Map_Array is array (Socket_Opt) of int;
Map : constant Map_Array :=
(AFFINITY => Low_Level.ZMQ_AFFINITY,
IDENTITY => Low_Level.ZMQ_IDENTITY,
SUBSCRIBE => Low_Level.ZMQ_SUBSCRIBE,
UNSUBSCRIBE => Low_Level.ZMQ_UNSUBSCRIBE,
RATE => Low_Level.ZMQ_RATE,
RECOVERY_IVL => Low_Level.ZMQ_RECOVERY_IVL,
SNDBUF => Low_Level.ZMQ_SNDBUF,
RCVBUF => Low_Level.ZMQ_RCVBUF,
RCVMORE => Low_Level.ZMQ_RCVMORE,
FD => Low_Level.ZMQ_FD,
EVENTS => Low_Level.ZMQ_EVENTS,
GET_TYPE => Low_Level.ZMQ_TYPE,
LINGER => Low_Level.ZMQ_LINGER,
RECONNECT_IVL => Low_Level.ZMQ_RECONNECT_IVL,
BACKLOG => Low_Level.ZMQ_BACKLOG,
RECONNECT_IVL_MAX => Low_Level.ZMQ_RECONNECT_IVL_MAX,
MAXMSGSIZE => Low_Level.ZMQ_MAXMSGSIZE,
SNDHWM => Low_Level.ZMQ_SNDHWM,
RCVHWM => Low_Level.ZMQ_RCVHWM,
MULTICAST_HOPS => Low_Level.ZMQ_MULTICAST_HOPS,
RCVTIMEO => Low_Level.ZMQ_RCVTIMEO,
SNDTIMEO => Low_Level.ZMQ_SNDTIMEO,
IPV4ONLY => Low_Level.ZMQ_IPV4ONLY,
LAST_ENDPOINT => Low_Level.ZMQ_LAST_ENDPOINT,
ROUTER_BEHAVIOR => Low_Level.ZMQ_ROUTER_BEHAVIOR,
TCP_KEEPALIVE => Low_Level.ZMQ_TCP_KEEPALIVE,
TCP_KEEPALIVE_CNT => Low_Level.ZMQ_TCP_KEEPALIVE_CNT,
TCP_KEEPALIVE_IDLE => Low_Level.ZMQ_TCP_KEEPALIVE_IDLE,
TCP_KEEPALIVE_INTVL => Low_Level.ZMQ_TCP_KEEPALIVE_INTVL,
TCP_ACCEPT_FILTER => Low_Level.ZMQ_TCP_ACCEPT_FILTER,
DELAY_ATTACH_ON_CONNECT => Low_Level.ZMQ_DELAY_ATTACH_ON_CONNECT,
XPUB_VERBOSE => Low_Level.ZMQ_XPUB_VERBOSE
(ZMQ_TYPE => Low_Level.ZMQ_TYPE,
AFFINITY => Low_Level.ZMQ_AFFINITY,
IDENTITY => Low_Level.ZMQ_IDENTITY,
SUBSCRIBE => Low_Level.ZMQ_SUBSCRIBE,
UNSUBSCRIBE => Low_Level.ZMQ_UNSUBSCRIBE,
RATE => Low_Level.ZMQ_RATE,
RECOVERY_IVL => Low_Level.ZMQ_RECOVERY_IVL,
SNDBUF => Low_Level.ZMQ_SNDBUF,
RCVBUF => Low_Level.ZMQ_RCVBUF,
RCVMORE => Low_Level.ZMQ_RCVMORE,
FD => Low_Level.ZMQ_FD,
EVENTS => Low_Level.ZMQ_EVENTS,
GET_TYPE => Low_Level.ZMQ_TYPE,
LINGER => Low_Level.ZMQ_LINGER,
RECONNECT_IVL => Low_Level.ZMQ_RECONNECT_IVL,
BACKLOG => Low_Level.ZMQ_BACKLOG,
RECONNECT_IVL_MAX => Low_Level.ZMQ_RECONNECT_IVL_MAX,
MAXMSGSIZE => Low_Level.ZMQ_MAXMSGSIZE,
SNDHWM => Low_Level.ZMQ_SNDHWM,
RCVHWM => Low_Level.ZMQ_RCVHWM,
MULTICAST_HOPS => Low_Level.ZMQ_MULTICAST_HOPS,
RCVTIMEO => Low_Level.ZMQ_RCVTIMEO,
SNDTIMEO => Low_Level.ZMQ_SNDTIMEO,
IPV4ONLY => Low_Level.ZMQ_IPV4ONLY,
LAST_ENDPOINT => Low_Level.ZMQ_LAST_ENDPOINT,
ROUTER_MANDATORY => Low_Level.ZMQ_ROUTER_MANDATORY,
TCP_KEEPALIVE => Low_Level.ZMQ_TCP_KEEPALIVE,
TCP_KEEPALIVE_CNT => Low_Level.ZMQ_TCP_KEEPALIVE_CNT,
TCP_KEEPALIVE_IDLE => Low_Level.ZMQ_TCP_KEEPALIVE_IDLE,
TCP_KEEPALIVE_INTVL => Low_Level.ZMQ_TCP_KEEPALIVE_INTVL,
TCP_ACCEPT_FILTER => Low_Level.ZMQ_TCP_ACCEPT_FILTER,
DELAY_ATTACH_ON_CONNECT => Low_Level.ZMQ_DELAY_ATTACH_ON_CONNECT,
XPUB_VERBOSE => Low_Level.ZMQ_XPUB_VERBOSE
);
function Img (Item : Ada.Streams.Stream_Element_Array) return String is
Ret : String (1 .. Item'Length * 2);
Cursor : Natural := 1;
type Map_String is array (Stream_Element (0) ..
Stream_Element (15)) of Character;
Hex : constant Map_String := ('0', '1', '2', '3',
'4', '5', '6', '7',
'8', '9', 'A', 'B',
'C', 'D', 'E', 'F');
begin
for I in Item'Range loop
Ret (Cursor) := Hex (Item (I) / 16);
Cursor := Cursor + 1;
Ret (Cursor) := Hex (Item (I) mod 16);
Cursor := Cursor + 1;
end loop;
return Ret;
end Img;
----------------
-- Initialize --
----------------
overriding procedure Initialize
(This : in out Socket)
not overriding procedure Initialize
(This : in out Socket;
With_Context : Contexts.Context;
Kind : Socket_Type)
is
begin
This.C := Low_Level.zmq_socket (This.With_Context.GetImpl,
Socket_Type'Pos (This.Kind));
This.C := Low_Level.zmq_socket (With_Context.GetImpl,
Socket_Type'Pos (Kind));
if This.C = Null_Address then
raise ZMQ_Error with "Unable to initialize";
end if;
@ -272,21 +258,35 @@ package body ZMQ.Sockets is
not overriding
procedure Send (This : in out Socket;
Msg_Addres : System.Address;
Msg_Length : Natural;
Flags : Socket_Flags := No_Flags) is
procedure Send (This : in out Socket;
Msg_Address : System.Address;
Msg_Length : Natural;
Flags : Socket_Flags := No_Flags) is
Ret : int;
begin
Ret := Low_Level.zmq_send (This.C, Msg_Addres,
Ret := Low_Level.zmq_send (This.C, Msg_Address,
size_t (Msg_Length),
int (Flags));
if Ret /= 0 then
if Ret = -1 then
raise ZMQ_Error with Error_Message (GNAT.OS_Lib.Errno) & " in " &
GNAT.Source_Info.Enclosing_Entity;
end if;
end Send;
not overriding procedure Send
(This : in out Socket;
Msg : ZMQ.Messages.Message'Class;
Flags : Socket_Flags := No_Flags) is
Ret : int;
begin
Ret := Low_Level.zmq_msg_send (Msg.GetImpl, This.C,
int (Flags));
if Ret = -1 then
raise ZMQ_Error with Error_Message (GNAT.OS_Lib.Errno) & " in " &
GNAT.Source_Info.Enclosing_Entity;
end if;
end Send;
-- -----------
-- -- flush --
@ -310,16 +310,16 @@ package body ZMQ.Sockets is
not overriding procedure Recv
(This : in Socket;
Msg : Messages.Message'Class;
Msg : in out Messages.Message'Class;
Flags : Socket_Flags := No_Flags)
is
Ret : int;
begin
Ret := Low_Level.zmq_recvmsg (This.C,
Msg.GetImpl,
int (Flags));
Ret := Low_Level.zmq_msg_recv (Msg.GetImpl,
This.C,
int (Flags));
if Ret /= 0 then
if Ret = -1 then
raise ZMQ_Error with Error_Message (GNAT.OS_Lib.Errno) & " in "
& GNAT.Source_Info.Enclosing_Entity;
end if;
@ -339,17 +339,28 @@ package body ZMQ.Sockets is
function Recv (This : in Socket;
Max_Length : Natural := 1024;
Flags : Socket_Flags := No_Flags) return String is
Buffer : String (1 .. Max_Length);
-- Buffer : String (1 .. Max_Length);
begin
This.Recv (Buffer'Address, Buffer'Length, Flags);
-- This.Recv (Buffer'Address, Buffer'Length, Flags);
raise Program_Error with "function Recv not implemented";
return "dummy";
end Recv;
procedure Recv (This : in Socket;
Msg : out Ada.Strings.Unbounded.Unbounded_String;
Flags : Socket_Flags := No_Flags) is
Temp_Msg : Messages.Message;
begin
Msg := Ada.Strings.Unbounded.To_Unbounded_String (This.Recv (Flags));
Temp_Msg.Initialize;
This.Recv (Temp_Msg, Flags);
declare
type Msg_Str is new String (1 .. Temp_Msg.GetSize);
package Conv is new System.Address_To_Access_Conversions (Msg_Str);
begin
Set_Unbounded_String
(Msg, String (Conv.To_Pointer (Temp_Msg.GetData).all));
end;
Temp_Msg.Finalize;
end Recv;
@ -382,13 +393,31 @@ package body ZMQ.Sockets is
end Finalize;
not overriding
procedure Set_High_Water_Mark_For_Outbound_Messages
(This : in out Socket;
Messages : Positive := 1000) is
begin
This.Setsockopt (Option => SNDHWM,
Value => Messages);
end Set_High_Water_Mark_For_Outbound_Messages;
not overriding
procedure Set_High_Water_Mark_For_Inbound_Messages
(This : in out Socket;
Messages : Positive := 1000) is
begin
This.Setsockopt (Option => RCVHWM,
Value => Messages);
end Set_High_Water_Mark_For_Inbound_Messages;
not overriding
procedure Set_IO_Thread_Affinity (This : in out Socket;
Value : Natural) is
procedure Set_IO_Thread_Affinity (This : in out Socket;
Value : Thread_Bitmap) is
begin
This.Setsockopt (AFFINITY, Value);
This.Setsockopt (AFFINITY, Value'Address, (Thread_Bitmap'Size + 1) / 8);
end Set_IO_Thread_Affinity;
not overriding
@ -406,29 +435,36 @@ package body ZMQ.Sockets is
This.Setsockopt (IDENTITY, Value);
end Set_Socket_Identity;
not overriding
procedure Set_Socket_Identity
(This : in out Socket;
Value : Ada.Strings.Unbounded.Unbounded_String) is
begin
This.Set_Socket_Identity (Ada.Strings.Unbounded.To_String (Value));
end Set_Socket_Identity;
not overriding
procedure Establish_Message_Filter (This : in out Socket;
procedure Set_Message_Filter (This : in out Socket;
Value : String) is
begin
This.Setsockopt (SUBSCRIBE, Value);
end Establish_Message_Filter;
end Set_Message_Filter;
not overriding
procedure Establish_Message_Filter
procedure Set_Message_Filter
(This : in out Socket;
Value : Ada.Streams.Stream_Element_Array) is
begin
This.Setsockopt (SUBSCRIBE, Value);
end Establish_Message_Filter;
end Set_Message_Filter;
procedure Establish_Message_Filter
procedure Set_Message_Filter
(This : in out Socket;
Value : Ada.Strings.Unbounded.Unbounded_String) is
begin
This.Setsockopt (SUBSCRIBE, To_String (Value));
end Establish_Message_Filter;
end Set_Message_Filter;
not overriding
@ -453,39 +489,296 @@ package body ZMQ.Sockets is
end Remove_Message_Filter;
not overriding
procedure Set_Multicast_Data_Rate (This : in out Socket;
Value : Natural) is
procedure Set_Multicast_Data_Rate
(This : in out Socket;
Kilobits_Per_Second : Natural := 100) is
begin
This.Setsockopt (RATE, Value);
This.Setsockopt (RATE, Kilobits_Per_Second);
end Set_Multicast_Data_Rate;
not overriding
procedure Set_Multicast_Recovery_Interval (This : in out Socket;
Value : Duration) is
procedure Set_Multicast_Recovery_Interval (This : in out Socket;
Interval : Duration) is
begin
This.Setsockopt (RECOVERY_IVL, Integer (Value));
This.Setsockopt (RECOVERY_IVL, Integer (Interval));
end Set_Multicast_Recovery_Interval;
not overriding
procedure Set_Kernel_Transmit_Buffer_Size (This : in out Socket;
Value : Natural) is
procedure Set_Kernel_Transmit_Buffer_Size (This : in out Socket;
Size : Natural) is
begin
This.Setsockopt (SNDBUF, Value);
This.Setsockopt (SNDBUF, Size);
end Set_Kernel_Transmit_Buffer_Size;
not overriding
procedure Set_Kernel_Receive_Buffer_Size (This : in out Socket;
Value : Natural) is
procedure Set_Kernel_Receive_Buffer_Size (This : in out Socket;
Size : Natural) is
begin
This.Setsockopt (RCVBUF, Value);
This.Setsockopt (RCVBUF, Size);
end Set_Kernel_Receive_Buffer_Size;
--<<
not overriding
procedure Set_Linger_Period_For_Socket_Shutdown
(This : in out Socket;
Period : Duration) is
begin
This.Set_Linger_Period_For_Socket_Shutdown (Integer (Period * 1000.0));
end Set_Linger_Period_For_Socket_Shutdown;
not overriding
procedure Set_Linger_Period_For_Socket_Shutdown
(This : in out Socket;
Period : Ada.Real_Time.Time_Span) is
begin
This.Set_Linger_Period_For_Socket_Shutdown
(Ada.Real_Time.To_Duration (Period));
end Set_Linger_Period_For_Socket_Shutdown;
not overriding
procedure Set_Linger_Period_For_Socket_Shutdown
(This : in out Socket;
Miliseconds : Integer) is
begin
This.Setsockopt (Option => LINGER,
Value => Integer'Max (-1, Miliseconds));
end Set_Linger_Period_For_Socket_Shutdown;
procedure Set_Reconnection_Interval
(This : in out Socket;
Interval : Duration) is
begin
This.Set_Reconnection_Interval (Integer (Interval * 1000.0));
end Set_Reconnection_Interval;
not overriding
procedure Set_Reconnection_Interval
(This : in out Socket;
Interval : Ada.Real_Time.Time_Span) is
begin
This.Set_Reconnection_Interval
(Ada.Real_Time.To_Duration (Interval));
end Set_Reconnection_Interval;
not overriding
procedure Set_Reconnection_Interval
(This : in out Socket;
Miliseconds : Integer) is
begin
This.Setsockopt (Option => RECONNECT_IVL,
Value => Integer'Max (-1, Miliseconds));
end Set_Reconnection_Interval;
procedure Set_Maximum_Reconnection_Interval
(This : in out Socket;
Interval : Duration := 0.0) is
begin
This.Set_Maximum_Reconnection_Interval (Integer (Interval * 1000.0));
end Set_Maximum_Reconnection_Interval;
not overriding
procedure Set_Maximum_Reconnection_Interval
(This : in out Socket;
Interval : Ada.Real_Time.Time_Span := Ada.Real_Time.Time_Span_Zero) is
begin
This.Set_Maximum_Reconnection_Interval
(Ada.Real_Time.To_Duration (Interval));
end Set_Maximum_Reconnection_Interval;
not overriding
procedure Set_Maximum_Reconnection_Interval
(This : in out Socket;
Miliseconds : Integer := 0) is
begin
This.Setsockopt (Option => RECONNECT_IVL_MAX,
Value => Integer'Max (-1, Miliseconds));
end Set_Maximum_Reconnection_Interval;
procedure Set_Maximum_Queue_Length_Of_Outstanding_Connections
(This : in out Socket;
Connections : Positive) is
begin
This.Setsockopt (Option => BACKLOG,
Value => Connections);
end Set_Maximum_Queue_Length_Of_Outstanding_Connections;
procedure Set_Maximum_Acceptable_Inbound_Message_Size
(This : in out Socket;
Size : Integer) is
begin
This.Setsockopt (Option => MAXMSGSIZE,
Value => Integer'Max (-1, Size));
end Set_Maximum_Acceptable_Inbound_Message_Size;
procedure Set_Maximum_Network_Hops_For_Multicast_Packets
(This : in out Socket;
Max_Hops : Positive := 1)
is
begin
This.Setsockopt (Option => MULTICAST_HOPS,
Value => Max_Hops);
end Set_Maximum_Network_Hops_For_Multicast_Packets;
procedure Set_Recieve_Time_Out
(This : in out Socket;
Time : Duration) is
begin
This.Set_Recieve_Time_Out (Integer (Time * 1000.0));
end Set_Recieve_Time_Out;
procedure Set_Recieve_Time_Out
(This : in out Socket;
Time : Ada.Real_Time.Time_Span) is
begin
This.Set_Recieve_Time_Out (Ada.Real_Time.To_Duration (Time));
end Set_Recieve_Time_Out;
procedure Set_Recieve_Time_Out
(This : in out Socket;
Milliseconds : Integer) is
begin
This.Setsockopt (Option => RCVTIMEO,
Value => Integer'Max (-1, Milliseconds));
end Set_Recieve_Time_Out;
procedure Set_Send_Time_Out
(This : in out Socket;
Time : Duration) is
begin
This.Set_Send_Time_Out (Integer (Time * 1000.0));
end Set_Send_Time_Out;
procedure Set_Send_Time_Out
(This : in out Socket;
Time : Ada.Real_Time.Time_Span) is
begin
This.Set_Send_Time_Out (Ada.Real_Time.To_Duration (Time));
end Set_Send_Time_Out;
procedure Set_Send_Time_Out
(This : in out Socket;
Milliseconds : Integer) is
begin
This.Setsockopt (Option => SNDTIMEO,
Value => Integer'Max (-1, Milliseconds));
end Set_Send_Time_Out;
not overriding
procedure Use_IPv4_Sockets_Only
(This : in out Socket;
Value : Boolean) is
begin
This.Setsockopt (Option => IPV4ONLY,
Value => Value);
end Use_IPv4_Sockets_Only;
not overriding
procedure Accept_Messages_Only_When_Connections_Are_Made
(This : in out Socket;
Value : Boolean) is
begin
This.Setsockopt (Option => DELAY_ATTACH_ON_CONNECT,
Value => Value);
end Accept_Messages_Only_When_Connections_Are_Made;
procedure Set_Accept_Only_Routable_Messages_On_ROUTER_Sockets
(This : in out Socket;
Value : Boolean) is
begin
This.Setsockopt (Option => ROUTER_MANDATORY,
Value => Value);
end Set_Accept_Only_Routable_Messages_On_ROUTER_Sockets;
not overriding
procedure Provide_All_Subscription_Messages_On_XPUB_Sockets
(This : in out Socket;
Value : Boolean) is
begin
This.Setsockopt (Option => XPUB_VERBOSE,
Value => Value);
end Provide_All_Subscription_Messages_On_XPUB_Sockets;
not overriding
procedure Override_SO_KEEPALIVE_Socket_Option
(This : in out Socket;
Value : SO_KEEPALIVE_Type) is
Map : constant array (SO_KEEPALIVE_Type) of Integer := (-1, 0, 1);
begin
This.Setsockopt (Option => TCP_KEEPALIVE,
Value => Map (Value));
end Override_SO_KEEPALIVE_Socket_Option;
not overriding
procedure Override_TCP_KEEPCNT_IDLE_Socket_Option
(This : in out Socket;
Value : Integer := -1) is
begin
if Value < -1 or Value = 0 then
raise Constraint_Error with "Value out of bounds:" & Value'Img;
end if;
This.Setsockopt (Option => TCP_KEEPALIVE_IDLE,
Value => Value);
end Override_TCP_KEEPCNT_IDLE_Socket_Option;
not overriding
procedure Override_TCP_KEEPCNT_CNT_Socket_Option
(This : in out Socket;
Value : Integer := -1) is
begin
if Value < -1 or Value = 0 then
raise Constraint_Error with "Value out of bounds:" & Value'Img;
end if;
This.Setsockopt (Option => TCP_KEEPALIVE_CNT,
Value => Value);
end Override_TCP_KEEPCNT_CNT_Socket_Option;
not overriding
procedure Override_TCP_KEEPINTVL_socket_option
(This : in out Socket;
Value : Integer := -1) is
begin
if Value < -1 or Value = 0 then
raise Constraint_Error with "Value out of bounds:" & Value'Img;
end if;
This.Setsockopt (Option => TCP_KEEPALIVE_INTVL,
Value => Value);
end Override_TCP_KEEPINTVL_socket_option;
not overriding
procedure Assign_Filters_To_Allow_New_TCP_Connections
(This : in out Socket;
Filter : String) is
begin
This.Setsockopt (Option => TCP_ACCEPT_FILTER,
Value => Filter);
end Assign_Filters_To_Allow_New_TCP_Connections;
not overriding
procedure Assign_Filters_To_Allow_New_TCP_Connections
(This : in out Socket;
Filter : Ada.Strings.Unbounded.Unbounded_String) is
begin
This.Setsockopt (Option => TCP_ACCEPT_FILTER,
Value => To_String (Filter));
end Assign_Filters_To_Allow_New_TCP_Connections;
not overriding
procedure Assign_Filters_To_Allow_New_TCP_Connections
(This : in out Socket;
Filter : Ada.Streams.Stream_Element_Array) is
begin
This.Setsockopt (Option => TCP_ACCEPT_FILTER,
Value => Filter);
end Assign_Filters_To_Allow_New_TCP_Connections;
--=======================================================================
--=======================================================================
function Get_Impl (This : in Socket) return System.Address is
begin
return This.C;
end Get_Impl;
-------------
--=======================================================================
--=======================================================================
not overriding
procedure Getsockopt (This : in Socket;
@ -550,7 +843,6 @@ package body ZMQ.Sockets is
end return;
end Getsockopt;
not overriding
function Getsockopt
(This : in Socket;
Option : Socket_Opt) return Ada.Streams.Stream_Element_Array is
@ -563,6 +855,13 @@ package body ZMQ.Sockets is
end Getsockopt;
not overriding
function Retrieve_Socket_Type (This : Socket) return Socket_Type is
begin
return Socket_Type'Val (Natural'(This.Getsockopt (ZMQ_TYPE)));
end Retrieve_Socket_Type;
not overriding
function More_Message_Parts_To_Follow (This : Socket) return Boolean is
begin
return Ret : Boolean do
@ -570,6 +869,21 @@ package body ZMQ.Sockets is
end return;
end More_Message_Parts_To_Follow;
not overriding
function Get_High_Water_Mark_For_Outbound_Messages
(This : Socket) return Natural is
begin
return This.Getsockopt (SNDHWM);
end Get_High_Water_Mark_For_Outbound_Messages;
not overriding
function Get_High_Water_Mark_For_Inbound_Messages
(This : Socket) return Natural is
begin
return This.Getsockopt (RCVHWM);
end Get_High_Water_Mark_For_Inbound_Messages;
function Get_IO_Thread_Affinity (This : Socket) return Thread_Bitmap is
Value_Size : Natural;
begin
@ -586,6 +900,21 @@ package body ZMQ.Sockets is
begin
return This.Getsockopt (IDENTITY);
end Get_Socket_Identity;
not overriding
function Get_Socket_Identity
(This : Socket)
return String is
begin
return This.Getsockopt (IDENTITY);
end Get_Socket_Identity;
not overriding
function Get_Socket_Identity
(This : Socket)
return Ada.Strings.Unbounded.Unbounded_String is
begin
return To_Unbounded_String (String'(This.Getsockopt (IDENTITY)));
end Get_Socket_Identity;
function Get_Multicast_Data_Rate (This : Socket) return Natural is
begin
@ -607,4 +936,224 @@ package body ZMQ.Sockets is
return This.Getsockopt (RCVBUF);
end Get_Kernel_Receive_Buffer_Size;
not overriding
function Get_Linger_Period_For_Socket_Shutdown
(This : Socket) return Duration is
begin
return Natural'(This.Get_Linger_Period_For_Socket_Shutdown) * 1000.0;
end Get_Linger_Period_For_Socket_Shutdown;
not overriding
function Get_Linger_Period_For_Socket_Shutdown
(This : Socket) return Ada.Real_Time.Time_Span is
begin
return Ada.Real_Time.To_Time_Span
(This.Get_Linger_Period_For_Socket_Shutdown);
end Get_Linger_Period_For_Socket_Shutdown;
not overriding
function Get_Linger_Period_For_Socket_Shutdown
(This : Socket) return Natural is
begin
return This.Getsockopt (LINGER);
end Get_Linger_Period_For_Socket_Shutdown;
not overriding
function Get_Reconnection_Interval
(This : Socket) return Duration is
begin
return Natural'(This.Get_Reconnection_Interval) * 1000.0;
end Get_Reconnection_Interval;
not overriding
function Get_Reconnection_Interval
(This : Socket) return Ada.Real_Time.Time_Span is
begin
return Ada.Real_Time.To_Time_Span
(This.Get_Reconnection_Interval);
end Get_Reconnection_Interval;
not overriding
function Get_Reconnection_Interval
(This : Socket) return Natural is
begin
return This.Getsockopt (RECONNECT_IVL);
end Get_Reconnection_Interval;
not overriding
function Get_Maximum_Reconnection_Interval
(This : Socket) return Duration is
begin
return Natural'(This.Get_Maximum_Reconnection_Interval) * 1000.0;
end Get_Maximum_Reconnection_Interval;
not overriding
function Get_Maximum_Reconnection_Interval
(This : Socket) return Ada.Real_Time.Time_Span is
begin
return Ada.Real_Time.To_Time_Span
(This.Get_Maximum_Reconnection_Interval);
end Get_Maximum_Reconnection_Interval;
not overriding
function Get_Maximum_Reconnection_Interval
(This : Socket) return Natural is
begin
return This.Getsockopt (RECONNECT_IVL_MAX);
end Get_Maximum_Reconnection_Interval;
not overriding
function Get_Maximum_Length_Of_The_Queue_Of_Outstanding_Connections
(This : Socket) return Natural is
begin
return This.Getsockopt (BACKLOG);
end Get_Maximum_Length_Of_The_Queue_Of_Outstanding_Connections;
not overriding
function Get_Maximum_Acceptable_Inbound_Message_Size
(This : Socket) return Integer is
begin
return This.Getsockopt (MAXMSGSIZE);
end Get_Maximum_Acceptable_Inbound_Message_Size;
not overriding
function Get_Maximum_Network_Hops_For_Multicast_Packets
(This : Socket) return Positive is
begin
return This.Getsockopt (MULTICAST_HOPS);
end Get_Maximum_Network_Hops_For_Multicast_Packets;
not overriding
function Get_Recieve_Timeout
(This : Socket) return Duration is
Temp : constant Integer := This.Get_Recieve_Timeout;
begin
if Temp < 0 then
return Duration'Last;
else
return 1000.0 * Temp;
end if;
end Get_Recieve_Timeout;
not overriding
function Get_Recieve_Timeout
(This : Socket) return Ada.Real_Time.Time_Span is
begin
return Ada.Real_Time.To_Time_Span (This.Get_Recieve_Timeout);
end Get_Recieve_Timeout;
not overriding
function Get_Recieve_Timeout -- Millisecond
(This : Socket) return Integer is
begin
return This.Getsockopt (RCVTIMEO);
end Get_Recieve_Timeout;
not overriding
function Get_Send_Timeout
(This : Socket) return Duration is
Temp : constant Integer := This.Get_Send_Timeout;
begin
if Temp < 0 then
return Duration'Last;
else
return 1000.0 * Temp;
end if;
end Get_Send_Timeout;
not overriding
function Get_Send_Timeout
(This : Socket) return Ada.Real_Time.Time_Span is
begin
return Ada.Real_Time.To_Time_Span (This.Get_Send_Timeout);
end Get_Send_Timeout;
not overriding
function Get_Send_Timeout -- Millisecond
(This : Socket) return Integer is
begin
return This.Getsockopt (SNDTIMEO);
end Get_Send_Timeout;
not overriding
function Get_IPv4_only_socket_override
(This : Socket) return Boolean is
begin
return This.Getsockopt (IPV4ONLY);
end Get_IPv4_only_socket_override;
not overriding
function Get_Attach_On_Connect
(This : Socket) return Boolean is
begin
return This.Getsockopt (DELAY_ATTACH_ON_CONNECT);
end Get_Attach_On_Connect;
not overriding
function Get_File_Descriptor
(This : Socket) return GNAT.OS_Lib.File_Descriptor is
begin
return GNAT.OS_Lib.File_Descriptor
(Interfaces.C.unsigned_long'(This.Getsockopt (FD)));
end Get_File_Descriptor;
not overriding
function Get_Last_Endpoint_Set
(This : Socket) return String is
begin
return This.Getsockopt (LAST_ENDPOINT);
end Get_Last_Endpoint_Set;
not overriding
function Get_Last_Endpoint_Set
(This : Socket) return Ada.Strings.Unbounded.Unbounded_String is
begin
return Ret : Ada.Strings.Unbounded.Unbounded_String do
Set_Unbounded_String (Ret, This.Get_Last_Endpoint_Set);
end return;
end Get_Last_Endpoint_Set;
not overriding
function Get_SO_KEEPALIVE_Socket_Option
(This : in Socket) return SO_KEEPALIVE_Type is
Map : constant array (-1 .. 1) of SO_KEEPALIVE_Type :=
(OS_Default, Disable, Enable);
begin
return Map (This.Getsockopt (LAST_ENDPOINT));
end Get_SO_KEEPALIVE_Socket_Option;
not overriding
function Get_TCP_KEEPCNT_IDLE_socket_option
(This : in Socket) return Integer is
begin
return This.Getsockopt (TCP_KEEPALIVE_IDLE);
end Get_TCP_KEEPCNT_IDLE_socket_option;
not overriding
function Get_TCP_KEEPCNT_CNT_socket_option
(This : in Socket) return Integer is
begin
return This.Getsockopt (TCP_KEEPALIVE_CNT);
end Get_TCP_KEEPCNT_CNT_socket_option;
not overriding
function Get_TCP_KEEPINTVL_Socket_Option
(This : in Socket) return Integer is
begin
return This.Getsockopt (TCP_KEEPALIVE_INTVL);
end Get_TCP_KEEPINTVL_Socket_Option;
procedure Set_Monitor
(This : Socket;
Monitor : Any_Socket_Monitor) is
begin
null;
end Set_Monitor;
end ZMQ.Sockets;

566
src/zmq-sockets.ads
View File

@ -36,7 +36,8 @@ with ZMQ.Contexts;
with ZMQ.Messages;
with System;
with GNAT.OS_Lib;
private with Interfaces.C;
with Interfaces.C;
with Ada.Real_Time;
package ZMQ.Sockets is
type Socket_Type is
@ -52,13 +53,11 @@ package ZMQ.Sockets is
XPUB,
XSUB);
type Socket
(With_Context : Contexts.Any_Context;
Kind : Socket_Type) is
type Socket is
new Ada.Finalization.Limited_Controlled with private;
type Any_Socket is access all Socket'Class;
Null_Socket : constant Socket;
type Socket_Flags is mod 2 ** 32;
@ -69,6 +68,11 @@ package ZMQ.Sockets is
More : constant Socket_Flags := 2#0000_0000_0000_0001#;
Shared : constant Socket_Flags := 2#0000_0000_1000_0000#;
not overriding
procedure Initialize
(This : in out Socket;
With_Context : Contexts.Context;
Kind : Socket_Type);
not overriding
procedure Bind
@ -80,11 +84,38 @@ package ZMQ.Sockets is
(This : in out Socket;
Address : Ada.Strings.Unbounded.Unbounded_String);
type Thread_Bitmap is array (0 .. 63) of Boolean;
pragma Pack (Thread_Bitmap);
not overriding
procedure Set_High_Water_Mark_For_Outbound_Messages
(This : in out Socket;
Messages : Positive := 1000);
-- Sets the high water mark for outbound messages on the specified socket.
-- The high water mark is a hard limit on the maximum number of outstanding
-- messages ØMQ shall queue in memory for any single peer that the
-- specified socket is communicating with.
-- If this limit has been reached the socket enters an exceptional state
-- and depending on the socket type, ØMQ will take appropriate action
-- such as blocking or dropping sent messages.
not overriding
procedure Set_High_Water_Mark_For_Inbound_Messages
(This : in out Socket;
Messages : Positive := 1000);
-- Sets the high water mark for inbound messages on the specified socket.
-- The high water mark is a hard limit on the maximum number of outstanding
-- messages ØMQ shall queue in memory for any single peer that the
-- specified socket is communicating with.
-- If this limit has been reached the socket enters an exceptional state
-- and depending on the socket type, ØMQ will take appropriate action
-- such as blocking or dropping sent messages.
not overriding
procedure Set_IO_Thread_Affinity
(This : in out Socket;
Value : Natural);
Value : Thread_Bitmap);
-- Sets the I/O thread affinity for newly created connections on the
-- specified socket.
-- Affinity determines which threads from the 0MQ I/O thread pool
@ -105,6 +136,10 @@ package ZMQ.Sockets is
(This : in out Socket;
Value : String);
not overriding
procedure Set_Socket_Identity
(This : in out Socket;
Value : Ada.Strings.Unbounded.Unbounded_String);
not overriding
procedure Set_Socket_Identity
(This : in out Socket;
Value : Ada.Streams.Stream_Element_Array);
@ -124,15 +159,15 @@ package ZMQ.Sockets is
-- by 0MQ infrastructure.
not overriding
procedure Establish_Message_Filter
procedure Set_Message_Filter
(This : in out Socket;
Value : String);
not overriding
procedure Establish_Message_Filter
procedure Set_Message_Filter
(This : in out Socket;
Value : Ada.Strings.Unbounded.Unbounded_String);
not overriding
procedure Establish_Message_Filter
procedure Set_Message_Filter
(This : in out Socket;
Value : Ada.Streams.Stream_Element_Array);
-- Establishes a new message filter on a SUB socket.
@ -168,15 +203,15 @@ package ZMQ.Sockets is
not overriding
procedure Set_Multicast_Data_Rate
(This : in out Socket;
Value : Natural);
(This : in out Socket;
Kilobits_Per_Second : Natural := 100);
-- Sets the maximum send or receive data rate for multicast transports
-- such as PGM using the specified socket.
not overriding
procedure Set_Multicast_Recovery_Interval
(This : in out Socket;
Value : Duration);
(This : in out Socket;
Interval : Duration);
-- Sets the recovery interval in seconds for multicast transports using
-- the specified socket.
-- The recovery interval determines the maximum time in seconds that a
@ -192,7 +227,7 @@ package ZMQ.Sockets is
not overriding
procedure Set_Kernel_Transmit_Buffer_Size
(This : in out Socket;
Value : Natural);
Size : Natural);
-- Sets the underlying kernel transmit buffer size for the socket
-- to the specified size in bytes.
-- A value of zero means leave the OS default unchanged.
@ -202,30 +237,270 @@ package ZMQ.Sockets is
not overriding
procedure Set_Kernel_Receive_Buffer_Size
(This : in out Socket;
Value : Natural);
Size : Natural);
-- Sets the underlying kernel receive buffer size for the socket to
-- the specified size in bytes.
-- A value of zero means leave the OS default unchanged.
-- For details refer to your operating system documentation for the
-- SO_RCVBUF socket option.
not overriding
procedure Set_Linger_Period_For_Socket_Shutdown
(This : in out Socket;
Period : Duration);
not overriding
procedure Set_Linger_Period_For_Socket_Shutdown
(This : in out Socket;
Period : Ada.Real_Time.Time_Span);
not overriding
procedure Set_Linger_Period_For_Socket_Shutdown
(This : in out Socket;
Miliseconds : Integer);
-- The linger period determines how long pending messages which have yet
-- to be sent to a peer shall linger in memory after a socket is closed.
-- and further affects the termination of the socket's context.
-- The following outlines the different behaviours:
-- The default value of 'first specifies an infinite linger period.
-- Pending messages shall not be discarded after a call to close;
-- attempting to terminate the socket's context blocks until all
-- pending messages have been sent to a peer.
-- The value of 0 specifies no linger period.
-- Pending messages shall be discarded immediately when the
-- socket is closed.
-- Positive values specify an upper bound for the linger period.
-- Pending messages shall not be discarded after a call to close();
-- attempting to terminate the socket's context blocks until either
-- all pending messages have been sent to a peer,
-- or the linger period expires, after which any pending
-- messages are discarded.
not overriding
procedure Set_Reconnection_Interval
(This : in out Socket;
Interval : Duration);
not overriding
procedure Set_Reconnection_Interval
(This : in out Socket;
Interval : Ada.Real_Time.Time_Span);
not overriding
procedure Set_Reconnection_Interval
(This : in out Socket;
Miliseconds : Integer);
-- Set the initial reconnection interval for the specified socket.
-- The reconnection interval is the period ØMQ shall wait between attempts
-- to reconnect disconnected peers when using connection-oriented
-- transports.
-- A negative value means no reconnection.
not overriding
procedure Set_Maximum_Reconnection_Interval
(This : in out Socket;
Interval : Duration := 0.0);
not overriding
procedure Set_Maximum_Reconnection_Interval
(This : in out Socket;
Interval : Ada.Real_Time.Time_Span := Ada.Real_Time.Time_Span_Zero);
not overriding
procedure Set_Maximum_Reconnection_Interval
(This : in out Socket;
Miliseconds : Integer := 0);
-- Set the maximum reconnection interval for the specified socket.
-- This is the maximum period ØMQ shall wait between attempts to reconnect.
-- On each reconnect attempt, the previous interval shall be doubled untill
-- Maximum_Reconnection_Interval is reached.
-- This allows for exponential backoff strategy.
-- Default value means no exponential backoff is performed and reconnect
-- interval calculations are only based on Reconnection_Interval.
not overriding
procedure Set_Maximum_Queue_Length_Of_Outstanding_Connections
(This : in out Socket;
Connections : Positive);
-- Set the maximum length of the queue of outstanding peer connections
-- for the specified socket;
-- this only applies to connection-oriented transports.
-- For details refer to your operating system documentation for the
-- listen function.
not overriding
procedure Set_Maximum_Acceptable_Inbound_Message_Size
(This : in out Socket;
Size : Integer);
-- Limits the size of the inbound message.
-- If a peer sends a message larger than SIZE it is disconnected.
-- A negative means no limit.
not overriding
procedure Set_Maximum_Network_Hops_For_Multicast_Packets
(This : in out Socket;
Max_Hops : Positive := 1);
-- Sets the time-to-live field in every multicast packet sent
-- from this socket. The default is 1 which means that the
-- multicast packets don't leave the local network.
not overriding
procedure Set_Recieve_Time_Out
(This : in out Socket;
Time : Duration);
not overriding
procedure Set_Recieve_Time_Out
(This : in out Socket;
Time : Ada.Real_Time.Time_Span);
not overriding
procedure Set_Recieve_Time_Out
(This : in out Socket;
Milliseconds : Integer);
-- Sets the timeout for receive operation on the socket.
-- If the value is 0, recv will fail immediately,
-- with a EAGAIN error if there is no message to receive.
-- If the value is Negative, it will block until a message is available.
-- For all other values, it will wait for a message for that amount of time
-- before Failing with an EAGAIN error.
not overriding
procedure Set_Send_Time_Out
(This : in out Socket;
Time : Duration);
not overriding
procedure Set_Send_Time_Out
(This : in out Socket;
Time : Ada.Real_Time.Time_Span);
not overriding
procedure Set_Send_Time_Out
(This : in out Socket;
Milliseconds : Integer);
-- Sets the timeout for send operation on the socket.
-- If the value is zero, send will fail immediately, with a EAGAIN error
-- if the message cannot be sent.
-- If the value is negative, it will block until the message is sent.
-- For all other values, it will try to send the message for that amount
-- of time before failing with an EAGAIN error.
not overriding
procedure Use_IPv4_Sockets_Only
(This : in out Socket;
Value : Boolean);
-- Sets the underlying native socket type.
-- If set to True will use IPv4 sockets, while the value of False
-- will use IPv6 sockets.
-- An IPv6 socket lets applications connect to and accept connections
-- from both IPv4 and IPv6 hosts.
not overriding
procedure Accept_Messages_Only_When_Connections_Are_Made
(This : in out Socket;
Value : Boolean);
-- If set , will delay the attachment of a pipe on connect until
-- the underlying connection has completed.
-- This will cause the socket to block if there are no other connections,
-- but will prevent queues from filling on pipes awaiting connection.
not overriding
procedure Set_Accept_Only_Routable_Messages_On_ROUTER_Sockets
(This : in out Socket;
Value : Boolean);
-- Sets the ROUTER socket behavior when an unroutable message is
-- encountered. A value of False is the default and discards the
-- message silently when it cannot be routed.
-- A value of True Raises EHOSTUNREACH if the message cannot be routed.
not overriding
procedure Provide_All_Subscription_Messages_On_XPUB_Sockets
(This : in out Socket;
Value : Boolean);
-- Sets the XPUB socket behavior on new subscriptions and unsubscriptions.
-- A value of False is the default and passes only new
-- subscription messages to upstream.
-- A value of True passes all subscription messages upstream.
type SO_KEEPALIVE_Type is (OS_Default, Disable, Enable);
not overriding
procedure Override_SO_KEEPALIVE_Socket_Option
(This : in out Socket;
Value : SO_KEEPALIVE_Type);
-- Override TCP_KEEPCNT(or TCP_KEEPALIVE on some OS) socket option
-- (where supported by OS).
not overriding
procedure Override_TCP_KEEPCNT_IDLE_Socket_Option
(This : in out Socket;
Value : Integer := -1);
-- Override TCP_KEEPCNT socket option(where supported by OS).
-- The default value of -1 means to skip any overrides and leave it
-- to OS default.
not overriding
procedure Override_TCP_KEEPCNT_CNT_Socket_Option
(This : in out Socket;
Value : Integer := -1);
-- Override TCP_KEEPCNT(or TCP_KEEPALIVE on some OS)
-- socket option(where supported by OS).
-- The default value of -1 means to skip any overrides and leave it to
-- OS default.
not overriding
procedure Override_TCP_KEEPINTVL_socket_option
(This : in out Socket;
Value : Integer := -1);
-- Override TCP_KEEPINTVL socket option(where supported by OS).
-- The default value of -1 means to skip any overrides and leave it to
-- OS default.
not overriding
procedure Assign_Filters_To_Allow_New_TCP_Connections
(This : in out Socket;
Filter : String);
not overriding
procedure Assign_Filters_To_Allow_New_TCP_Connections
(This : in out Socket;
Filter : Ada.Strings.Unbounded.Unbounded_String);
not overriding
procedure Assign_Filters_To_Allow_New_TCP_Connections
(This : in out Socket;
Filter : Ada.Streams.Stream_Element_Array);
-- Assign arbitrary number of filters that will be applied for
-- each new TCP transport connection on a listening socket.
-- If no filters applied, then TCP transport allows connections from
-- any ip.
-- If at least one filter is applied then new connection source ip
-- should be matched. To clear all filters call
-- Assign_Filters_To_Allow_New_TCP_Connections(socket, "").
-- Filter is a null-terminated string with ipv6 or ipv4 CIDR.
--=======================================================================
--=======================================================================
not overriding
function Retrieve_Socket_Type (This : Socket) return Socket_Type;
-- Retrieve the socket type for the specified socket.
-- The socket type is specified at socket creation time and
-- cannot be modified afterwards.
not overriding
function More_Message_Parts_To_Follow (This : Socket) return Boolean;
-- Returns True if the multi-part message currently being read from the
-- specified socket has more message parts to follow.
-- If there are no message parts to follow or if the message currently
-- being read is not a multi-part message a value of True will be returned.
-- Otherwise, False will be returned.
-- returns True if the message part last received from the socket was
-- a data part with more parts to follow.
not overriding
function Get_High_Water_Mark_For_Outbound_Messages
(This : Socket) return Natural;
-- Returns the high water mark for outbound messages on the
-- specified socket.
type Thread_Bitmap is array (0 .. 63) of Boolean;
pragma Pack (Thread_Bitmap);
not overriding
function Get_High_Water_Mark_For_Inbound_Messages
(This : Socket) return Natural;
-- Return the high water mark for inbound messages on the specified socket.
not overriding
function Get_IO_Thread_Affinity (This : Socket) return Thread_Bitmap;
-- Returns the I/O thread affinity for newly created connections
-- on the specified socket.
@ -239,6 +514,15 @@ package ZMQ.Sockets is
-- a value of 3 specifies that subsequent connections on socket shall be
-- handled exclusively by I/O threads 1 and 2.
not overriding
function Get_Socket_Identity
(This : Socket)
return String;
not overriding
function Get_Socket_Identity
(This : Socket)
return Ada.Strings.Unbounded.Unbounded_String;
not overriding
function Get_Socket_Identity
(This : Socket)
return Ada.Streams.Stream_Element_Array;
@ -257,10 +541,12 @@ package ZMQ.Sockets is
-- Identities starting with binary zero are reserved for use by the
-- ZMQ infrastructure.
not overriding
function Get_Multicast_Data_Rate (This : Socket) return Natural;
-- Returns the maximum send or receive data rate for multicast transports
-- using the specified socket.
not overriding
function Get_Multicast_Recovery_Interval (This : Socket) return Duration;
-- Retrieves the recovery interval for multicast transports using the
-- specified socket.
@ -268,6 +554,7 @@ package ZMQ.Sockets is
-- a receiver can be absent from a multicast group before unrecoverable
-- data loss will occur.
not overriding
function Get_Kernel_Transmit_Buffer_Size (This : Socket) return Natural;
-- Returns the underlying kernel transmit buffer size for the
-- specified socket.
@ -275,6 +562,7 @@ package ZMQ.Sockets is
-- For details refer to your operating system documentation for
-- the SO_SNDBUF socket option.
not overriding
function Get_Kernel_Receive_Buffer_Size (This : Socket) return Natural;
-- Returns the underlying kernel receive buffer size for the
-- specified socket.
@ -282,37 +570,167 @@ package ZMQ.Sockets is
-- For details refer to your operating system documentation
-- for the SO_RCVBUF socket option
not overriding
function Get_Linger_Period_For_Socket_Shutdown
(This : Socket) return Duration;
not overriding
function Get_Linger_Period_For_Socket_Shutdown
(This : Socket) return Ada.Real_Time.Time_Span;
not overriding
function Get_Linger_Period_For_Socket_Shutdown -- Millisecond
(This : Socket) return Natural;
-- Retrieves the linger period for the specified socket.
not overriding procedure Connect
not overriding
function Get_Reconnection_Interval
(This : Socket) return Duration;
not overriding
function Get_Reconnection_Interval
(This : Socket) return Ada.Real_Time.Time_Span;
not overriding
function Get_Reconnection_Interval -- Millisecond
(This : Socket) return Natural;
-- Retrieves the initial reconnection interval for the specified socket.
not overriding
function Get_Maximum_Reconnection_Interval
(This : Socket) return Duration;
not overriding
function Get_Maximum_Reconnection_Interval
(This : Socket) return Ada.Real_Time.Time_Span;
not overriding
function Get_Maximum_Reconnection_Interval -- Millisecond
(This : Socket) return Natural;
-- Retrieves the maximum reconnection interval for the specified socket.
not overriding
function Get_Maximum_Length_Of_The_Queue_Of_Outstanding_Connections
(This : Socket) return Natural;
-- Retrieve the maximum length of the queue of outstanding peer connections
-- for the specified socket;
-- this only applies to connection-oriented transports.
not overriding
function Get_Maximum_Acceptable_Inbound_Message_Size
(This : Socket) return Integer;
-- Retrieves limit for the inbound messages.
not overriding
function Get_Maximum_Network_Hops_For_Multicast_Packets
(This : Socket) return Positive;
-- Retrieves time-to-live used for outbound multicast packets.
not overriding
function Get_Recieve_Timeout
(This : Socket) return Duration;
not overriding
function Get_Recieve_Timeout
(This : Socket) return Ada.Real_Time.Time_Span;
not overriding
function Get_Recieve_Timeout -- Millisecond
(This : Socket) return Integer;
-- Retrieves the timeout for recv operation on the socket.
not overriding
function Get_Send_Timeout
(This : Socket) return Duration;
not overriding
function Get_Send_Timeout
(This : Socket) return Ada.Real_Time.Time_Span;
not overriding
function Get_Send_Timeout -- Millisecond
(This : Socket) return Integer;
-- Retrieves the timeout for send operation on the socket.
not overriding
function Get_IPv4_only_socket_override
(This : Socket) return Boolean;
-- Retrives the underlying native socket type.
not overriding
function Get_Attach_On_Connect
(This : Socket) return Boolean;
-- Retrieves the state of the attach on connect value.
not overriding
function Get_File_Descriptor
(This : Socket) return GNAT.OS_Lib.File_Descriptor;
-- Retrieves the file descriptor associated with the specified socket.
-- The returned file descriptor can be used to integrate the socket
-- into an existing event loop;
-- the ØMQ library shall signal any pending events on the socket
-- in an edge-triggered fashion by making the file descriptor become
-- ready for reading..
not overriding
function Get_Last_Endpoint_Set
(This : Socket) return String;
not overriding
function Get_Last_Endpoint_Set
(This : Socket) return Ada.Strings.Unbounded.Unbounded_String;
-- Retrieves the last endpoint bound for TCP and IPC transports.
-- The returned value will be a string in the form of a ZMQ DSN.
-- Note that if the TCP host is INADDR_ANY, indicated by a *,
-- then the returned address will be 0.0.0.0 (for IPv4).
not overriding
function Get_SO_KEEPALIVE_Socket_Option
(This : in Socket) return SO_KEEPALIVE_Type;
not overriding
function Get_TCP_KEEPCNT_IDLE_socket_option
(This : in Socket) return Integer;
not overriding
function Get_TCP_KEEPCNT_CNT_socket_option
(This : in Socket) return Integer;
not overriding
function Get_TCP_KEEPINTVL_Socket_Option
(This : in Socket) return Integer;
not overriding
procedure Connect
(This : in out Socket;
Address : String);
not overriding
procedure Connect
(This : in out Socket;
Address : Ada.Strings.Unbounded.Unbounded_String);
not overriding procedure Send
not overriding
procedure Send
(This : in out Socket;
Msg : String;
Flags : Socket_Flags := No_Flags);
not overriding procedure Send
not overriding
procedure Send
(This : in out Socket;
Msg : Ada.Strings.Unbounded.Unbounded_String;
Flags : Socket_Flags := No_Flags);
not overriding procedure Send
not overriding
procedure Send
(This : in out Socket;
Msg : Ada.Streams.Stream_Element_Array;
Flags : Socket_Flags := No_Flags);
not overriding procedure Send
(This : in out Socket;
Msg_Addres : System.Address;
Msg_Length : Natural;
Flags : Socket_Flags := No_Flags);
not overriding
procedure Send
(This : in out Socket;
Msg : ZMQ.Messages.Message'Class;
Flags : Socket_Flags := No_Flags);
not overriding
procedure Send
(This : in out Socket;
Msg_Address : System.Address;
Msg_Length : Natural;
Flags : Socket_Flags := No_Flags);
-- Queues the message referenced by the msg argument to be sent to socket
-- The flags argument is a combination of the flags defined below:
-- NOBLOCK
@ -348,41 +766,57 @@ package ZMQ.Sockets is
generic
type Element is private;
pragma Compile_Time_Error
(Element'Has_Access_Values, "No access values allowed in Element");
procedure Send_Generic
(This : in out Socket;
Msg : Element;
Flags : Socket_Flags := No_Flags);
-- not overriding
-- procedure flush (This : in out Socket);
-- generic
-- type Element is private;
-- with procedure Write (S : access Ada.Streams.Root_Stream_Type'Class;
-- Data : Element);
-- procedure Send_Indefinite_Generic
-- (This : in out Socket;
-- Msg : Element;
-- Flags : Socket_Flags := No_Flags);
-- not overriding
-- procedure flush (This : in out Socket);
not overriding
procedure Recv
(This : in Socket;
Msg : out Ada.Strings.Unbounded.Unbounded_String;
Flags : Socket_Flags := No_Flags);
not overriding function Recv
(This : in Socket;
Flags : Socket_Flags := No_Flags)
not overriding
function Recv
(This : in Socket;
Max_Length : Natural := 1024;
Flags : Socket_Flags := No_Flags)
return String;
not overriding function Recv
not overriding
function Recv
(This : in Socket;
Flags : Socket_Flags := No_Flags)
return Ada.Strings.Unbounded.Unbounded_String;
not overriding procedure Recv
not overriding
procedure Recv
(This : in Socket;
Msg : Messages.Message'Class;
Msg : in out Messages.Message'Class;
Flags : Socket_Flags := No_Flags);
procedure Recv (This : in Socket; Flags : Socket_Flags := No_Flags);
not overriding
procedure Recv
(This : in Socket;
Flags : Socket_Flags := No_Flags);
not overriding
procedure Close (This : in out Socket) renames Finalize;
--
type Socket_Monitor is limited interface;
@ -396,7 +830,7 @@ package ZMQ.Sockets is
Address : String;
Err : Integer) is null;
procedure Connect_Retried
(This : Socket_Monitor;
(This : Socket_Monitor;
Address : String;
Interval : Duration) is null;
procedure Listening
@ -428,27 +862,19 @@ package ZMQ.Sockets is
Address : String;
Fd : GNAT.OS_Lib.File_Descriptor) is null;
procedure Set_Monitor (This : Socket;
Monitor : Any_Socket_Monitor);
procedure Set_Monitor (This : Socket;
Monitor : Any_Socket_Monitor);
-- function "=" (Left, Right : in Context) return Boolean;
function Get_Impl (This : in Socket) return System.Address;
private
type Socket
(With_Context : Contexts.Any_Context;
Kind : Socket_Type)
is new Ada.Finalization.Limited_Controlled with record
C : System.Address := System.Null_Address;
end record;
function Img (Item : Ada.Streams.Stream_Element_Array) return String;
overriding
procedure Initialize
(This : in out Socket);
overriding
procedure Finalize (This : in out Socket);
type Socket_Opt is (AFFINITY,
--
-- Low level setopt getopt operations.
--
type Socket_Opt is (ZMQ_TYPE,
AFFINITY,
IDENTITY,
SUBSCRIBE,
UNSUBSCRIBE,
@ -472,7 +898,7 @@ private
SNDTIMEO,
IPV4ONLY,
LAST_ENDPOINT,
ROUTER_BEHAVIOR,
ROUTER_MANDATORY,
TCP_KEEPALIVE,
TCP_KEEPALIVE_CNT,
TCP_KEEPALIVE_IDLE,
@ -508,7 +934,6 @@ private
Value_Size : Natural);
--------------------------------------------------------
--------------------------------------------------------
function Getsockopt
(This : in Socket;
@ -538,6 +963,21 @@ private
Value : System.Address;
Value_Size : out Natural);
private
type Socket
is new Ada.Finalization.Limited_Controlled with record
C : System.Address := System.Null_Address;
end record;
overriding
procedure Finalize (This : in out Socket);
MAX_OPTION_SIZE : constant := 256;
Null_Socket : constant Socket :=
(Ada.Finalization.Limited_Controlled with
C => System.Null_Address);
end ZMQ.Sockets;

515
src/zmq-sockets.ads.npp
View File

@ -1,515 +0,0 @@
-------------------------------------------------------------------------------
-- --
-- 0MQ Ada-binding --
-- --
-- Z M Q . S O C K E T S --
-- --
-- S p e c --
-- --
-- Copyright (C) 2010-2011, per.sandberg@bredband.net --
-- --
-- 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. --
-------------------------------------------------------------------------------
with Ada.Streams;
with Ada.Finalization;
with Ada.Strings.Unbounded;
with ZMQ.Messages;
with ZMQ.Contexts;
with System;
private with Interfaces.C;
package ZMQ.Sockets is
type Socket_Type is
(PAIR,
PUB,
SUB,
REQ,
REP,
XREQ,
XREP,
PULL,
PUSH);
type Socket is new Ada.Finalization.Limited_Controlled with private;
type Socket_Flags is mod 2 ** 32;
pragma Warnings (Off);
function "+" (L, R : Socket_Flags) return Socket_Flags renames "or";
pragma Warnings (On);
No_Flags : constant Socket_Flags := 2#0000_0000_0000_0000#;
More : constant Socket_Flags := 2#0000_0000_0000_0001#;
Shared : constant Socket_Flags := 2#0000_0000_1000_0000#;
not overriding
procedure Initialize (This : in out Socket;
With_Context : Contexts.Context;
Kind : Socket_Type);
not overriding
procedure Bind (This : in out Socket;
Address : String);
not overriding
procedure Bind (This : in out Socket;
Address : Ada.Strings.Unbounded.Unbounded_String);
not overriding
procedure Set_High_Water_Mark
(This : in out Socket;
Value : Natural);
-- Sets the high water mark for the specified socket.
-- The high water mark is a hard limit on the maximum number of
-- outstanding messages 0MQ shall queue in memory for any single peer
-- that the specified socket is communicating with.
-- If this limit has been reached the socket shall enter an exceptional
-- state and depending on the socket type,
-- 0MQ shall take appropriate action such as blocking or dropping
-- sent messages.
-- Refer to the individual socket descriptions in zmq_socket(3)
-- for details on the exact action taken for each socket type.
-- The default ZMQ_HWM value of zero means "no limit".
not overriding
procedure Set_Disk_Offload_Size (This : in out Socket;
Value : Natural);
-- Sets the disk offload (swap) size < for the specified socket.
-- A socket which has ZMQ_SWAP set to a non - zero value may exceed
-- in this case outstanding messages shall be offloaded to storage on
-- disk rather than held in memory.
-- The value defines the maximum size of the swap space in bytes
not overriding
procedure Set_IO_Thread_Affinity (This : in out Socket;
Value : Natural);
-- Sets the I/O thread affinity for newly created connections on the
-- specified socket.
-- Affinity determines which threads from the 0MQ I/O thread pool
-- created connections.
-- A value of zero specifies no affinity, meaning that work shall be
-- distributed fairly among all 0MQ I/O threads in the thread pool.
-- For non-zero values,
-- the lowest bit corresponds to thread 1, second lowest bit to thread 2
-- and so on.
-- For example, a value of 3 specifies that subsequent connections on
-- socket shall behandled exclusively by I/O threads 1 and 2.
-- See also zmq_init(3) for details on allocating the number
-- of I/O threads for a specific context.
not overriding
procedure Set_Socket_Identity
(This : in out Socket;
Value : String);
not overriding
procedure Set_Socket_Identity
(This : in out Socket;
Value : Ada.Streams.Stream_Element_Array);
-- Sets the identity of the specified socket.
-- Socket identity determines if existing 0MQ infastructure
-- (message queues, forwarding devices) shall be identified with a specific
-- application and persist across multiple runs of the application.
-- If the socket has no identity, each run of an application is completely
-- separate from other runs. However, with identity set the socket shall
-- re-use any existing 0MQ infrastructure configured by the
-- previous run(s).
-- Thus the application may receive messages that were sent in the
-- meantime, message queue limits shall be shared with previous run(s)
-- and so on.
-- Identity should be at least one byte and at most 255 bytes long.
-- Identities starting with binary zero are reserved for use
-- by 0MQ infrastructure.
not overriding
procedure Establish_Message_Filter (This : in out Socket;
Value : String);
not overriding
procedure Establish_Message_Filter
(This : in out Socket;
Value : Ada.Strings.Unbounded.Unbounded_String);
procedure Establish_Message_Filter
(This : in out Socket;
Value : Ada.Streams.Stream_Element_Array);
-- Establishes a new message filter on a SUB socket.
-- Newly created SUB sockets filters out all incoming messages,
-- therefore you should call this option to establish an initial
-- message filter.
-- An empty option_value of length zero shall subscribe to all
-- incoming messages.
-- A non-empty option_value shall subscribe to all messages beginning
-- with the specified prefix.
-- Mutiple filters may be attached to a single SUB socket,
-- in which case a message shall be accepted
-- if it matches at least one filter.
not overriding
procedure Remove_Message_Filter (This : in out Socket;
Value : String);
not overriding
procedure Remove_Message_Filter
(This : in out Socket;
Value : Ada.Strings.Unbounded.Unbounded_String);
not overriding
procedure Remove_Message_Filter
(This : in out Socket;
Value : Ada.Streams.Stream_Element_Array);
-- Remove an existing message filter on a SUB socket.
-- The filter specified must match an existing filter previously
-- established with "Establish_message_filter".
-- If the socket has several instances of the same filter attached the
-- Remove_message_filter removes only one instance,
-- leaving the rest in place and functional.
not overriding
procedure Set_Multicast_Data_Rate (This : in out Socket;
Value : Natural);
-- Sets the maximum send or receive data rate for multicast transports
-- such as PGM using the specified socket.
not overriding
procedure Set_Multicast_Recovery_Interval (This : in out Socket;
Value : Duration);
-- Sets the recovery interval in seconds for multicast transports using
-- the specified socket.
-- The recovery interval determines the maximum time in seconds that a
-- receiver can be absent from a multicast group before unrecoverable
-- data loss will occur.
-- Caution:
-- Excersize care when setting large recovery intervals as the data needed
-- for recovery will be held in memory.
-- For example, a 1 minute recovery interval at a data rate of
-- 1Gbps requires a 7GB in-memory buffer.
not overriding
procedure Set_Multicast_Loopback (This : in out Socket;
Enable : Boolean);
-- Controls whether data sent via multicast transports using
-- the specified socket can also be received by the sending host
-- via loopback.
-- A value of False disables the loopback functionality,
-- while the default value of True enables the loopback functionality.
-- Leaving multicast loopback enabled when it is not required can have
-- a negative impact on performance.
-- Where possible, disable multicast_loopback
-- in production environments.
not overriding
procedure Set_Kernel_Transmit_Buffer_Size (This : in out Socket;
Value : Natural);
-- Sets the underlying kernel transmit buffer size for the socket
-- to the specified size in bytes.
-- A value of zero means leave the OS default unchanged.
-- For details please refer to your operating system documentation
-- for the SO_SNDBUF socket option.
not overriding
procedure Set_Kernel_Receive_Buffer_Size (This : in out Socket;
Value : Natural);
-- Sets the underlying kernel receive buffer size for the socket to
-- the specified size in bytes.
-- A value of zero means leave the OS default unchanged.
-- For details refer to your operating system documentation for the
-- SO_RCVBUF socket option.
function More_Message_Parts_To_Follow (This : Socket) return Boolean;
-- Returns True if the multi-part message currently being read from the
-- specified socket has more message parts to follow.
-- If there are no message parts to follow or if the message currently
-- being read is not a multi-part message a value of True will be returned.
-- Otherwise, False will be returned.
function Get_High_Water_Mark (This : Socket) return Natural;
-- Returns the high water mark for the specified socket.
-- The high water mark is a hard limit on the maximum number of outstanding
-- messages ZMQ shall queue in memory for any single peer that
-- the specified socket is communicating with.
-- If this limit has been reached the socket shall enter an exceptional
-- state and depending on the socket type, ZMQ shall take appropriate
-- action such as blocking or dropping sent messages.
-- The default high_water_mark value of zero means "no limit".
function Get_Disk_Offload_Size (This : Socket) return Natural;
-- Returns the disk offload (swap) size for the specified socket.
-- A socket which has SWAP set to a non-zero value may exceed
-- in this case outstanding messages shall be offloaded to storage on disk
-- rather than held in memory.
-- The value of defines the maximum size of the swap space in bytes.
type Thread_Bitmap is array (0 .. 63) of Boolean;
pragma Pack (Thread_Bitmap);
function Get_IO_Thread_Affinity (This : Socket) return Thread_Bitmap;
-- Returns the I/O thread affinity for newly created connections
-- on the specified socket.
-- Affinity determines which threads from the ZMQ I/O thread pool
-- created connections.
-- A value of zero specifies no affinity, meaning that work shall be
-- distributed fairly among all ZMQ I/O threads in the thread pool.
-- For non-zero values, the lowest bit corresponds to thread 1,
-- second lowest bit to thread 2 and so on. For example,
-- a value of 3 specifies that subsequent connections on socket shall be
-- handled exclusively by I/O threads 1 and 2.
function Get_Socket_Identity
(This : Socket) return Ada.Streams.Stream_Element_Array;
-- Returns the identity of the specified socket.
-- Socket identity determines if existing ZMQ infastructure
-- (message queues, forwarding devices) shall be identified with a specific
-- application and persist across multiple runs of the application.
-- If the socket has no identity, each run of an application is completely
-- separate from other runs. However, with identity set the socket shall
-- re-use any existing ZMQ infrastructure configured by the
-- previous run(s).
-- Thus the application may receive messages that were sent
-- in the meantime,
-- message queue limits shall be shared with previous run(s) and so on.
-- Identity can be at least one byte and at most 255 bytes long.
-- Identities starting with binary zero are reserved for use by the
-- ZMQ infrastructure.
function Get_Multicast_Data_Rate (This : Socket) return Natural;
-- Returns the maximum send or receive data rate for multicast transports
-- using the specified socket.
function Get_Multicast_Recovery_Interval (This : Socket) return Duration;
-- Retrieves the recovery interval for multicast transports using the
-- specified socket.
-- The recovery interval determines the maximum time in seconds that
-- a receiver can be absent from a multicast group before unrecoverable
-- data loss will occur.
function Get_Multicast_Loopback (This : Socket) return Boolean;
-- Returns True if multicast transports shall be recievd bye the
-- loopback interface.
function Get_Kernel_Transmit_Buffer_Size (This : Socket) return Natural;
-- Returns the underlying kernel transmit buffer size for the
-- specified socket.
-- A value of zero means that the OS default is in effect.
-- For details refer to your operating system documentation for
-- the SO_SNDBUF socket option.
function Get_Kernel_Receive_Buffer_Size (This : Socket) return Natural;
-- Returns the underlying kernel receive buffer size for the
-- specified socket.
-- A value of zero means that the OS default is in effect.
-- For details refer to your operating system documentation
-- for the SO_RCVBUF socket option
not overriding
procedure Connect (This : in out Socket;
Address : String);
procedure Connect (This : in out Socket;
Address : Ada.Strings.Unbounded.Unbounded_String);
not overriding
procedure Send (This : in out Socket;
Msg : Messages.Message'Class;
Flags : Socket_Flags := No_Flags);
not overriding
procedure Send (This : in out Socket;
Msg : String;
Flags : Socket_Flags := No_Flags);
not overriding
procedure Send (This : in out Socket;
Msg : Ada.Strings.Unbounded.Unbounded_String;
Flags : Socket_Flags := No_Flags);
not overriding
procedure Send (This : in out Socket;
Msg : Ada.Streams.Stream_Element_Array;
Flags : Socket_Flags := No_Flags);
not overriding
procedure Send (This : in out Socket;
Msg_Addres : System.Address;
Msg_Length : Natural;
Flags : Socket_Flags := No_Flags);
-- Queues the message referenced by the msg argument to be sent to socket
-- The flags argument is a combination of the flags defined below:
-- NOBLOCK
-- Specifies that the operation should be performed in non-blocking mode.
-- If the message cannot be queued on the socket,
-- the send function shall fail with errno set to EAGAIN.
-- SNDMORE
-- Specifies that the message being sent is a multi-part message,
-- and that further message parts are to follow.
-- Refer to the section regarding multi-part messages
-- below for a detailed description.
-- Note!
-- A successful invocation of send does not indicate that the message
-- has been transmitted to the network,
-- only that it has been queued on the socket and 0MQ has assumed
-- responsibility for the message.
-- Multi-part messages
-- A 0MQ message is composed of 1 or more message parts;
-- each message part is an independent zmq_msg_t in its own right.
-- 0MQ ensures atomic delivery of messages;
-- peers shall receive either all message parts of
-- a message or none at all.
-- The total number of message parts is unlimited.
--
-- An application wishing to send a multi-part message does so by
-- specifying the SNDMORE flag to send.
-- The presence of this flag indicates to 0MQ that the message being sent
-- is a multi-part message and that more message parts are to follow.
-- When the application wishes to send the final message part it does so
-- by calling zmq without the SNDMORE flag;
-- this indicates that no more message parts are to follow.
-- Creates a Message and sends it over the socket.
generic
type Element is private;
procedure Send_Generic (This : in out Socket;
Msg : Element;
Flags : Socket_Flags := No_Flags);
-- not overriding
-- procedure flush (This : in out Socket);
not overriding
procedure Recv (This : in Socket;
Msg : Messages.Message'Class;
Flags : Socket_Flags := No_Flags);
procedure Recv (This : in Socket;
Msg : out Ada.Strings.Unbounded.Unbounded_String;
Flags : Socket_Flags := No_Flags);
not overriding
function Recv (This : in Socket;
Flags : Socket_Flags := No_Flags) return String;
not overriding
function Recv (This : in Socket;
Flags : Socket_Flags := No_Flags)
return Ada.Strings.Unbounded.Unbounded_String;
procedure Recv (This : in Socket;
Flags : Socket_Flags := No_Flags);
overriding
procedure Finalize (This : in out Socket);
procedure Close (This : in out Socket) renames Finalize;
--
-- function "=" (Left, Right : in Context) return Boolean;
function Get_Impl (This : in Socket) return System.Address;
private
type Socket is new Ada.Finalization.Limited_Controlled with record
C : System.Address := System.Null_Address;
end record;
function Img (Item : Ada.Streams.Stream_Element_Array) return String;
type Socket_Opt is
(HWM,
SWAP,
AFFINITY,
IDENTITY,
SUBSCRIBE,
UNSUBSCRIBE,
RATE,
RECOVERY_IVL,
MCAST_LOOP,
SNDBUF,
RCVBUF,
RCVMORE,
FD,
EVENTS,
GET_TYPE,
LINGER,
RECONNECT_IVL,
BACKLOG);
not overriding
procedure Setsockopt (This : in out Socket;
Option : Socket_Opt;
Value : String);
not overriding
procedure Setsockopt (This : in out Socket;
Option : Socket_Opt;
Value : Boolean);
not overriding
procedure Setsockopt (This : in out Socket;
Option : Socket_Opt;
Value : Natural);
not overriding
procedure Setsockopt
(This : in out Socket;
Option : Socket_Opt;
Value : Ada.Streams.Stream_Element_Array);
not overriding
procedure Setsockopt (This : in out Socket;
Option : Socket_Opt;
Value : System.Address;
Value_Size : Natural);
--------------------------------------------------------
--------------------------------------------------------
function Getsockopt (This : in Socket;
Option : Socket_Opt) return String;
not overriding
function Getsockopt (This : in Socket;
Option : Socket_Opt) return Boolean;
not overriding
function Getsockopt (This : in Socket;
Option : Socket_Opt) return Natural;
not overriding
function Getsockopt
(This : in Socket;
Option : Socket_Opt) return Interfaces.C.Unsigned_Long;
not overriding
function Getsockopt
(This : in Socket;
Option : Socket_Opt) return Ada.Streams.Stream_Element_Array;
not overriding
procedure Getsockopt (This : in Socket;
Option : Socket_Opt;
Value : System.Address;
Value_Size : out Natural);
MAX_OPTION_SIZE : constant := 256;
end ZMQ.Sockets;

508
src/zmq-utilities-memory_streams.adb
View File

@ -1,255 +1,253 @@
-------------------------------------------------------------------------------
-- --
-- Copyright 2007 Per Sandberg <per.sandberg@bredband.net> --
-- --
-- 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. --
-------------------------------------------------------------------------------
-- --
------------------------------------------------------------------------------
with Ada.IO_Exceptions;
with GNAT.Memory_Dump;
with System.Memory;
package body ZMQ.Utilities.Memory_Streams is
use Ada.Streams;
overriding
procedure Dump
(This : in Memory_Stream;
Full_Buffer : in Boolean := False;
Outf : in Text_IO.File_Access := Text_IO.Standard_Output) is
pragma Unreferenced (Outf);
Buffer : Large_Buffer_Access renames This.Buffer.As_Pointer;
begin
if Full_Buffer then
GNAT.Memory_Dump.Dump
(Buffer.all (Buffer.all'First)'Address,
Integer (This.Buffer_Length));
else
GNAT.Memory_Dump.Dump
(Buffer.all (Buffer.all'First)'Address,
Integer (This.Cursor) - 1);
end if;
end Dump;
overriding function Eof (This : in Memory_Stream) return Boolean is
begin
return This.Cursor > This.Buffer_Length;
end Eof;
-----------------
-- Get_Address --
-----------------
overriding
function Get_Address (This : in Memory_Stream) return System.Address is
begin
return This.Buffer.As_Address;
end Get_Address;
----------------
-- Get_Length --
----------------
overriding function Get_Length (This : in Memory_Stream)
return Ada.Streams.Stream_Element_Count is
begin
return This.Buffer_Length;
end Get_Length;
overriding procedure Seek (This : in out Memory_Stream;
Pos : in Ada.Streams.Stream_Element_Offset) is
begin
This.Cursor := This.Cursor + Pos;
end Seek;
overriding function Pos (This : in Memory_Stream)
return Ada.Streams.Stream_Element_Offset is
begin
return This.Cursor;
end Pos;
-----------------
-- Set_Address --
-----------------
overriding
procedure Set_Address
(This : in out Memory_Stream; To : in System.Address) is
begin
This.Buffer.As_Address := To;
end Set_Address;
----------------
-- Set_Length --
----------------
overriding
procedure Set_Length (This : in out Memory_Stream;
To : in Ada.Streams.Stream_Element_Count) is
begin
This.Buffer_Length := To;
This.Reset;
end Set_Length;
----------
-- Read --
----------
overriding
procedure Read
(This : in out Memory_Stream;
Item : out Stream_Element_Array;
Last : out Stream_Element_Offset)
is
First : Stream_Element_Offset;
LLast : Stream_Element_Offset;
begin
First := This.Cursor;
LLast := This.Cursor + Item'Length - 1;
if LLast > This.Buffer_Length then
raise Ada.IO_Exceptions.End_Error;
end if;
Item := This.Buffer.As_Pointer.all (First .. LLast);
This.Cursor := LLast + 1;
Last := Item'Last;
end Read;
-----------
-- Write --
-----------
overriding
procedure Write
(This : in out Memory_Stream;
Item : in Stream_Element_Array)
is
First : Stream_Element_Offset;
Last : Stream_Element_Offset;
begin
First := This.Cursor;
Last := This.Cursor + Item'Length - 1;
if Last > This.Buffer_Length then
raise Ada.IO_Exceptions.Device_Error;
end if;
This.Cursor := Last + 1;
This.Buffer.As_Pointer.all (First .. Last) := Item;
end Write;
overriding
procedure Reset (This : in out Memory_Stream) is
begin
This.Cursor := This.Buffer.As_Pointer.all'First;
end Reset;
procedure Read
(This : not null access Ada.Streams.Root_Stream_Type'Class;
Item : out Memory_Stream) is
pragma Unreferenced (This, Item);
begin
raise Program_Error with
"Its not possible to read into a memory stream using 'read";
end Read;
procedure Write
(This : not null access Ada.Streams.Root_Stream_Type'Class;
Item : in Memory_Stream) is
begin
Ada.Streams.Stream_Element_Array'Write
(This,
Item.Buffer.As_Pointer.all
(Item.Buffer.As_Pointer.all'First .. Item.Cursor));
end Write;
procedure Read
(This : not null access Ada.Streams.Root_Stream_Type'Class;
Item : out Dynamic_Memory_Stream) is
begin
Read (This, Memory_Stream (Item));
end Read;
procedure Write
(This : not null access Ada.Streams.Root_Stream_Type'Class;
Item : in Dynamic_Memory_Stream) is
begin
Write (This, Memory_Stream (Item));
end Write;
procedure Write
(This : in out Dynamic_Memory_Stream;
Item : in Ada.Streams.Stream_Element_Array) is
begin
if This.Cursor + Item'Length > This.Buffer_Length then
This.Expand (This.Cursor + Item'Length);
end if;
Memory_Stream (This).Write (Item);
end Write;
procedure Expand
(This : in out Dynamic_Memory_Stream;
to_Size : Ada.Streams.Stream_Element_Offset) is
new_Size : System.Memory.size_t := 0;
use System.Memory;
begin
while new_Size < size_t (to_Size) loop
case This.Strategy is
when As_Needed =>
new_Size := size_t (to_Size);
when Multiply_By_Two =>
new_Size := size_t (2 * This.Buffer_Length);
when Add_Initial_Size =>
new_Size := size_t (This.Buffer_Length + This.Initial_Size);
end case;
end loop;
This.Buffer.As_Address := System.Memory.Realloc
(This.Buffer.As_Address, new_Size);
This.Buffer_Length := Streams.Stream_Element_Count (new_Size);
end Expand;
procedure Initialize (This : in out Dynamic_Memory_Stream) is
use System.Memory;
begin
This.Buffer.As_Address :=
System.Memory.Alloc (size_t (This.Initial_Size));
This.Buffer_Length := This.Initial_Size;
end Initialize;
procedure Finalize (This : in out Dynamic_Memory_Stream) is
use System.Memory;
begin
System.Memory.Free (This.Buffer.As_Address);
end Finalize;
procedure Initialize (This : in out controler) is
use System.Memory;
begin
This.controled.Initialize;
end Initialize;
procedure Finalize (This : in out controler) is
begin
This.controled.Finalize;
end Finalize;
end ZMQ.Utilities.Memory_Streams;
-------------------------------------------------------------------------------
-- --
-- Copyright 2007 Per Sandberg <per.sandberg@bredband.net> --
-- --
-- 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. --
-------------------------------------------------------------------------------
-- --
------------------------------------------------------------------------------
with Ada.IO_Exceptions;
with GNAT.Memory_Dump;
with System.Memory;
package body ZMQ.Utilities.Memory_Streams is
use Ada.Streams;
overriding
procedure Dump
(This : in Memory_Stream;
Full_Buffer : in Boolean := False) is
Buffer : Large_Buffer_Access renames This.Buffer.As_Pointer;
begin
if Full_Buffer then
GNAT.Memory_Dump.Dump
(Buffer.all (Buffer.all'First)'Address,
Integer (This.Buffer_Length));
else
GNAT.Memory_Dump.Dump
(Buffer.all (Buffer.all'First)'Address,
Integer (This.Cursor) - 1);
end if;
end Dump;
overriding function Eof (This : in Memory_Stream) return Boolean is
begin
return This.Cursor > This.Buffer_Length;
end Eof;
-----------------
-- Get_Address --
-----------------
overriding
function Get_Address (This : in Memory_Stream) return System.Address is
begin
return This.Buffer.As_Address;
end Get_Address;
----------------
-- Get_Length --
----------------
overriding function Get_Length (This : in Memory_Stream)
return Ada.Streams.Stream_Element_Count is
begin
return This.Buffer_Length;
end Get_Length;
overriding procedure Seek (This : in out Memory_Stream;
Pos : in Ada.Streams.Stream_Element_Offset) is
begin
This.Cursor := This.Cursor + Pos;
end Seek;
overriding function Pos (This : in Memory_Stream)
return Ada.Streams.Stream_Element_Offset is
begin
return This.Cursor;
end Pos;
-----------------
-- Set_Address --
-----------------
overriding
procedure Set_Address
(This : in out Memory_Stream; To : in System.Address) is
begin
This.Buffer.As_Address := To;
end Set_Address;
----------------
-- Set_Length --
----------------
overriding
procedure Set_Length (This : in out Memory_Stream;
To : in Ada.Streams.Stream_Element_Count) is
begin
This.Buffer_Length := To;
This.Reset;
end Set_Length;
----------
-- Read --
----------
overriding
procedure Read
(This : in out Memory_Stream;
Item : out Stream_Element_Array;
Last : out Stream_Element_Offset)
is
First : Stream_Element_Offset;
LLast : Stream_Element_Offset;
begin
First := This.Cursor;
LLast := This.Cursor + Item'Length - 1;
if LLast > This.Buffer_Length then
raise Ada.IO_Exceptions.End_Error;
end if;
Item := This.Buffer.As_Pointer.all (First .. LLast);
This.Cursor := LLast + 1;
Last := Item'Last;
end Read;
-----------
-- Write --
-----------
overriding
procedure Write
(This : in out Memory_Stream;
Item : in Stream_Element_Array)
is
First : Stream_Element_Offset;
Last : Stream_Element_Offset;
begin
First := This.Cursor;
Last := This.Cursor + Item'Length - 1;
if Last > This.Buffer_Length then
raise Ada.IO_Exceptions.Device_Error;
end if;
This.Cursor := Last + 1;
This.Buffer.As_Pointer.all (First .. Last) := Item;
end Write;
overriding
procedure Reset (This : in out Memory_Stream) is
begin
This.Cursor := This.Buffer.As_Pointer.all'First;
end Reset;
procedure Read
(This : not null access Ada.Streams.Root_Stream_Type'Class;
Item : out Memory_Stream) is
pragma Unreferenced (This, Item);
begin
raise Program_Error with
"Its not possible to read into a memory stream using 'read";
end Read;
procedure Write
(This : not null access Ada.Streams.Root_Stream_Type'Class;
Item : in Memory_Stream) is
begin
Ada.Streams.Stream_Element_Array'Write
(This,
Item.Buffer.As_Pointer.all
(Item.Buffer.As_Pointer.all'First .. Item.Cursor));
end Write;
procedure Read
(This : not null access Ada.Streams.Root_Stream_Type'Class;
Item : out Dynamic_Memory_Stream) is
begin
Read (This, Memory_Stream (Item));
end Read;
procedure Write
(This : not null access Ada.Streams.Root_Stream_Type'Class;
Item : in Dynamic_Memory_Stream) is
begin
Write (This, Memory_Stream (Item));
end Write;
procedure Write
(This : in out Dynamic_Memory_Stream;
Item : in Ada.Streams.Stream_Element_Array) is
begin
if This.Cursor + Item'Length > This.Buffer_Length then
This.Expand (This.Cursor + Item'Length);
end if;
Memory_Stream (This).Write (Item);
end Write;
procedure Expand
(This : in out Dynamic_Memory_Stream;
to_Size : Ada.Streams.Stream_Element_Offset) is
new_Size : System.Memory.size_t := 0;
use System.Memory;
begin
while new_Size < size_t (to_Size) loop
case This.Strategy is
when As_Needed =>
new_Size := size_t (to_Size);
when Multiply_By_Two =>
new_Size := size_t (2 * This.Buffer_Length);
when Add_Initial_Size =>
new_Size := size_t (This.Buffer_Length + This.Initial_Size);
end case;
end loop;
This.Buffer.As_Address := System.Memory.Realloc
(This.Buffer.As_Address, new_Size);
This.Buffer_Length := Streams.Stream_Element_Count (new_Size);
end Expand;
procedure Initialize (This : in out Dynamic_Memory_Stream) is
use System.Memory;
begin
This.Buffer.As_Address :=
System.Memory.Alloc (size_t (This.Initial_Size));
This.Buffer_Length := This.Initial_Size;
end Initialize;
procedure Finalize (This : in out Dynamic_Memory_Stream) is
use System.Memory;
begin
System.Memory.Free (This.Buffer.As_Address);
end Finalize;
procedure Initialize (This : in out controler) is
use System.Memory;
begin
This.controled.Initialize;
end Initialize;
procedure Finalize (This : in out controler) is
begin
This.controled.Finalize;
end Finalize;
end ZMQ.Utilities.Memory_Streams;

16
src/zmq-utilities-memory_streams.ads
View File

@ -26,7 +26,6 @@
with Ada.Streams;
with System;
with Ada.Text_IO;
with Ada.Finalization;
package ZMQ.Utilities.Memory_Streams is
use Ada;
@ -50,6 +49,15 @@ package ZMQ.Utilities.Memory_Streams is
type Any_Memory_Stream_Interface is access
all Memory_Stream_Interface'Class;
--
procedure Read
(This : in out Memory_Stream_Interface;
Item : out Ada.Streams.Stream_Element_Array;
Last : out Ada.Streams.Stream_Element_Offset) is abstract;
procedure Write
(This : in out Memory_Stream_Interface;
Item : in Ada.Streams.Stream_Element_Array) is abstract;
function Get_Address
(This : in Memory_Stream_Interface) return System.Address is abstract;
-- Returns the Address to the real buffer
@ -87,8 +95,7 @@ package ZMQ.Utilities.Memory_Streams is
procedure Dump
(This : in Memory_Stream_Interface;
Full_Buffer : in Boolean := False;
Outf : in Text_IO.File_Access := Text_IO.Standard_Output) is null;
Full_Buffer : in Boolean := False) is null;
-- Dumps the contents of the buffer from the first element
-- to the cursor.
@ -140,8 +147,7 @@ package ZMQ.Utilities.Memory_Streams is
overriding
procedure Dump
(This : in Memory_Stream;
Full_Buffer : in Boolean := False;
Outf : in Ada.Text_IO.File_Access := Ada.Text_IO.Standard_Output);
Full_Buffer : in Boolean := False);
-- Dumps the contents of the buffer from the first element
-- to the cursor.

23
src/zmq-utilities-stream_element_array_image.adb Normal file
View File

@ -0,0 +1,23 @@
function ZMQ.Utilities.Stream_Element_Array_Image
(Item : Ada.Streams.Stream_Element_Array)
return String
is
use Ada.Streams;
Cursor : Natural := 1;
type Map_String is array (Stream_Element (0) ..
Stream_Element (15)) of Character;
Hex : constant Map_String := ('0', '1', '2', '3',
'4', '5', '6', '7',
'8', '9', 'A', 'B',
'C', 'D', 'E', 'F');
begin
return Ret : String (1 .. Item'Length * 2) do
for I in Item'Range loop
Ret (Cursor) := Hex (Item (I) / 16);
Cursor := Cursor + 1;
Ret (Cursor) := Hex (Item (I) mod 16);
Cursor := Cursor + 1;
end loop;
end return;
end ZMQ.Utilities.Stream_Element_Array_Image;

3
src/zmq-utilities-stream_element_array_image.ads Normal file
View File

@ -0,0 +1,3 @@
with Ada.Streams;
function ZMQ.Utilities.Stream_Element_Array_Image
(Item : Ada.Streams.Stream_Element_Array) return String;

2
tests/testcases/zmq-tests-testcases-test_compile.adb
View File

@ -4,6 +4,8 @@ with ZMQ.Contexts;
with ZMQ.Sockets;
with ZMQ.Messages;
with ZMQ.Low_Level;
with ZMQ.Proxys;
with ZMQ.Devices;
package body ZMQ.Tests.Testcases.Test_Compile is
use AUnit;

22
tests/testcases/zmq-tests-testcases-test_pubsub.adb
View File

@ -21,20 +21,21 @@ package body ZMQ.Tests.Testcases.Test_Pubsub is
Test_Port : constant String := "inproc://pub-sub";
-------------------------
-- initialize
-------------------------
procedure Initialize (Test : in out AUnit.Test_Cases.Test_Case'Class) is
T : Test_Case renames Test_Case (Test);
begin
T.Ctx.Initialize;
T.Pub.Initialize (T.Ctx, Sockets.PUB);
T.Sub.Initialize (T.Ctx, Sockets.SUB);
T.Sub.Establish_Message_Filter ("");
T.Sub.Set_Message_Filter ("");
T.Sub.Bind ("inproc://pub-sub");
T.Pub.Connect ("inproc://pub-sub");
T.Sub.Bind (Test_Port);
T.Pub.Connect (Test_Port);
end Initialize;
-------------------------
@ -43,9 +44,19 @@ package body ZMQ.Tests.Testcases.Test_Pubsub is
procedure Send (Test : in out AUnit.Test_Cases.Test_Case'Class) is
T : Test_Case renames Test_Case (Test);
Msg : Ada.Strings.Unbounded.Unbounded_String;
task Rec is
entry Has_Data;
end Rec;
task body Rec is
begin
T.Sub.Recv (Msg);
accept Has_Data;
end Rec;
begin
delay 0.01;
T.Pub.Send (MSG_STRING);
T.Sub.Recv (Msg);
Rec.Has_Data;
Assert (Msg = MSG_STRING, "Error");
end Send;
@ -57,7 +68,6 @@ package body ZMQ.Tests.Testcases.Test_Pubsub is
begin
T.Pub.Finalize;
T.Sub.Finalize;
T.Ctx.Finalize;
end Finalize;
--------------------
-- Register_Tests --

2
tests/testcases/zmq-tests-testcases-test_reqresp.adb
View File

@ -27,7 +27,6 @@ package body ZMQ.Tests.Testcases.Test_REQRESP is
procedure Initialize (Test : in out AUnit.Test_Cases.Test_Case'Class) is
T : Test_Case renames Test_Case (Test);
begin
T.Ctx.Initialize;
T.Sub.Initialize (T.Ctx, Sockets.REP);
T.Sub.Bind ("inproc://req");
@ -60,7 +59,6 @@ package body ZMQ.Tests.Testcases.Test_REQRESP is
begin
T.Pub.Finalize;
T.Sub.Finalize;
T.Ctx.Finalize;
end Finalize;
--------------------
-- Register_Tests --