bin/posix-mq.rb

   1 #!/usr/bin/ruby
   2 # -*- encoding: binary -*-
   3 Encoding.default_external = Encoding::BINARY if defined?(Encoding)
   4 $stderr.sync = $stdout.sync = true
   5
   6 require 'posix_mq'
   7 require 'optparse'
   8
   9 commands = %w(create attr send receive wait unlink)
  10 usage = "Usage: MQUEUE=/name #{File.basename($0)} COMMAND " \
  11         "[options] [<arguments>]\n" \
  12         "COMMAND may be one of: #{commands.join(', ')}"
  13
  14 mqueue = ENV["MQUEUE"] or abort usage
  15 command = ARGV.shift or abort usage
  16 commands.include?(command) or abort usage
  17
  18 priority = nil
  19 timeout = nil
  20 mode = 0666
  21 oflags = IO::RDONLY
  22 mq_attr = nil
  23 nonblock = false
  24 command = command.to_sym
  25
  26 ARGV.options do |x|
  27   x.banner = usage.split(/\n/).first.gsub(/COMMAND/, command.to_s)
  28   x.separator ''
  29
  30   case command
  31   when :create
  32     oflags |= IO::CREAT
  33     x.on('-x', '--exclusive', "exclusive create") {
  34       oflags |= IO::EXCL
  35     }
  36     x.on('-m', '--mode=MODE', "octal file mode") { |i|
  37       mode = i.to_i(8)
  38     }
  39     x.on('-c', '--maxmsg=COUNT', Integer, "maximum number of messages") { |i|
  40       mq_attr ||= POSIX_MQ::Attr.new
  41       mq_attr.maxmsg = i
  42     }
  43     x.on('-s', '--msgsize=BYTES', Integer, "maximum size of message") { |i|
  44       mq_attr ||= POSIX_MQ::Attr.new
  45       mq_attr.msgsize = i
  46     }
  47   when :wait
  48     x.on('-t', '--timeout=SECONDS', Float, "timeout in seconds") { |f|
  49       timeout = f
  50     }
  51   when :send, :receive
  52     conflict = "timeout and nonblock are exclusive"
  53     x.on('-t', '--timeout=SECONDS', Float, "timeout in seconds") { |f|
  54       abort conflict if nonblock
  55       timeout = f
  56     }
  57     x.on('-n', '--nonblock', "nonblocking operation") {
  58       abort conflict if timeout
  59       nonblock = true
  60       oflags |= IO::NONBLOCK
  61     }
  62     if command == :send
  63       oflags = IO::WRONLY
  64       x.on('-p', '--priority=PRIO', Integer, "priority of message") { |i|
  65         priority = i
  66       }
  67     else
  68       x.on('-p', '--priority', "output priority of message to stderr") {
  69         priority = $stderr
  70       }
  71     end
  72   end
  73   x.on('-q', "quiet warnings and errors") { $stderr.reopen("/dev/null", "w") }
  74   x.on('-h', '--help', 'Show this help message.') { puts x; exit }
  75   x.parse!
  76 end
  77
  78 trap(:INT) { exit 130 }
  79
  80 unless command == :send || ARGV.empty?
  81   abort "#{command} accepts no arguments"
  82 end
  83
  84 begin
  85   if command == :create && mq_attr
  86     mq_attr.flags = mq_attr.curmsgs = 0
  87     mq_attr.msgsize && ! mq_attr.maxmsg and
  88       abort "--maxmsg must be set with --msgsize"
  89     mq_attr.maxmsg && ! mq_attr.msgsize and
  90       abort "--msgsize must be set with --maxmsg"
  91   elsif command == :unlink
  92     POSIX_MQ.unlink(mqueue)
  93     exit
  94   end
  95
  96   mq = POSIX_MQ.open(mqueue, oflags, mode, mq_attr)
  97   case command
  98   when :create
  99     exit
 100   when :receive
 101     buf, prio = mq.receive("", timeout)
 102     $stderr.syswrite("priority=#{prio}\n") if priority
 103     $stdout.syswrite(buf)
 104   when :send
 105     ARGV << $stdin.read if ARGV.empty?
 106     ARGV.each { |msg| mq.send(msg, priority, timeout) }
 107   when :attr
 108     mq_attr = mq.attr
 109     $stdout.syswrite(
 110       "flags=#{mq_attr.flags}\n" \
 111       "maxmsg=#{mq_attr.maxmsg}\n" \
 112       "msgsize=#{mq_attr.msgsize}\n" \
 113       "curmsgs=#{mq_attr.curmsgs}\n")
 114   when :wait
 115     trap(:USR1) { exit }
 116
 117     # we wouldn't get a notification if there were already messages
 118     exit if mq.attr.curmsgs > 0
 119     mq.notify = :USR1
 120     exit if mq.attr.curmsgs > 0 # avoid race condition
 121
 122     timeout.nil? ? sleep : sleep(timeout)
 123     exit 2 # timed out
 124   end
 125 rescue Errno::EEXIST
 126   abort "Queue exists"
 127 rescue Errno::ENOENT
 128   abort "Queue does not exist"
 129 rescue Errno::EMSGSIZE
 130   abort "Message too long"
 131 rescue Errno::EAGAIN
 132   abort(command == :send ? "Queue full" : "No messages available")
 133 rescue Errno::ETIMEDOUT
 134   warn "Operation timed out"
 135   exit 2
 136 rescue => e
 137   abort e.message
 138 end