bin/posix-mq-rb

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