1. MQTT가 IoT에서의 메시징 프로토콜로 각광을 받고 있다. 정말 그런지 가능성을 확인해 본다.
2. IoT 환경에서 디바이스와 유저, 유저와 유저가 효과적으로 메시지를 주고 받을 수 있는 솔류션을 찾아보고 싶다. 
   
   

• 우분투 리눅스 server 14.04
• Mosquitto MQTT Broker
• Ruby : MQTT 클라이언트를 만들기 위한 언어 
  
  

• 요청과 응답채널을 분리한다. : 요청을 HTTP로 하려고 한다. HTTP는 훌륭한 툴들도 많고 개발 방법들도 잘 정리되어있다. 분산 처리가 용이해서 대용량 시스템 구성이 쉽다는 것도 장점이다. 반면 응답을 동기화하기에는 좀 애매모호하다는 단점이 있다. websocket을 이용하거나 long polling을 이용하는 방법도 있겠는데, HTTP의 장점에 흠집을 내는 격이라서 사용하기가 망설여진다. 하여, HTTP는 요청에만 사용하고 응답은 따로 분리하기로 했다. 이런 모델들 요즘엔 많이 찾아볼 수 있다.
• 응답 채널을 MQTT로 한다. : 유저가 인증을 끝내고 IoT 인프라에 접속하면, 메시지 Push를 위한 private topic을 만든다. 기기에서 발생한 메시지는 IoT 인프라로 Publish하고, 이걸 유저가 subscribe하는 방식이다. 실제 구현에서는 MQTT broker 앞에 MQTT Proxy등을 놓는 등의 장치가 필요하다. 
  
  

• Topic을 만들어서, 여러 유저(혹은 기기가) subscribe 하게 한다. .. 하나의 토픽에 여러 유저 혹은 디바이스가 PUB/SUB 하게 하면 된다.
• 운영 상의 이유로 여러 유저에게 메시지를 보내야 할 수도 있다. 이를 위해서 몇 개의 Public 채널을 만들 수도 있다. 앱 업데이트 Topic, CS Topic, (동의한 유저를 대상으로 하는) 이벤트 Topic 등이 그것이다. 
  
  

1. Client는 API Server에 인증을 요청한다.
2. 인증성공하면 session을 발행한다.
3. Client가 Push 서비스를 받기 원한다면, MQTT Proxy에 연결 요청한다.
4. 연결 후 첫 번째 메시지로 인증확인을 위한 Auth request 메시지를 Publish 한다.
   • 인증을 위한 전용의 Auth Topic을 만든다. /service/auth/<user_session> 정도면 되겠다.
5. Auth request 메시지를 받은 프락시 서버는 API Server에 세션이 올바른지 확인 요청을 한다. Session validation check 과정이 되겠다.
   • 실패하면 연결을 끊는다. 실패 메시지는 Auth Topic로 전송하면 되겠다.
6. 연결 성공하면 메시지 topic에 subscribe를 요청한다.
7. 프락시 서버는 subscribe 요청이 올바른지 확인 후 요청을 MQTT 브로커로 넘긴다. 
   
   

1. Proxy의 bind port는 5000이다.
2. 브로커의 서비스 포트는 1883(MQTT 기본 포트)이다. 
   
   

require 'rubygems'
require 'mqtt'
require 'pp'

# MQTT::Packet에 get 메서드 추가
# topic, payload(유저 메시지)
class MQTT::Packet
    def topic 
        return @topic
    end
    def payload
        return @payload
    end
end

# MQTT::Proxy의 process_packets 메서드 재 정의
# 원래의 process_packets 는 packet만 넘겨주는데, 이 걸로는  
# 이 packet에 어떤 클라이언트의 것인지 확인할 수 없다.
# 해서 클라이언트를 구분할 수 있는 값(소켓 객체의 id)을 추가로 넘기게 했다.
class MQTT::Proxy
  def process_packets(client_socket,server_socket)
    loop do
      selected = IO.select([client_socket,server_socket], nil, nil, @select_timeout)
      if selected.nil?
        # Timeout
        raise "Timeout in select"
      else
        if selected[0].include?(client_socket)
          packet = MQTT::Packet.read(client_socket)
          packet = @client_filter.call(packet, client_socket.object_id) unless @client_filter.nil?
          # packet이 nil 이라면 인증 오류인 걸로 한다.          
          if packet.nil?
            raise "Authorization Error"
          else
            server_socket.write(packet)
          end
        elsif selected[0].include?(server_socket)
          packet = MQTT::Packet.read(server_socket)
          packet = @server_filter.call(packet) unless @server_filter.nil?
          unless packet.nil?
            client_socket.write(packet)
          end
        else
            logger.error "Problem with select: socket is neither server or client"
        end
      end
    end
  end
end

class MQTTProxy
    attr_reader :proxy
    attr_reader :session_info
    attr_reader :logger
    def initialize(args)
        @proxy=MQTT::Proxy.new args
        @session_info = Hash.new

        @logger = args[:logger]
        if @logger.nil?
            @logger = Logger.new(STDOUT)
            @logger.level = Logger::INFO
        end
    end

    # 유저 세션을 검사한다. 
    # 귀찮아서 123456이면 성공인 걸로 했다.
    def validationSession? session 
        # session을 가져오기 위한 연산이 들어간다.
        puts "Check #{session == "123456"}"
        return session == "123456"
    end

    def run
        # 클라이언트 요청을 읽어서 분석한다. 
        # MQTT::Proxy에서 읽은 클라이언트의 MQTT 패킷과 id를 매개변수로 연산을 한다.
        # id는 현재 유저 세션을 가리키는 key로 socket의 object_id를 이용했다.
        @proxy.client_filter = lambda { |packet, id|
            # connect 한 후 첫번째 메시지에 대해서 세션을 검사한다.
            if @session_info.has_key?(id)
                if @session_info[id] == 0
                    return nil if !validationSession? packet.payload
                    logger.info "MQTT::Proxy User authentication success"
                end
                @session_info[id] += 1 
                @session_info['topic'] = "/user/#{id}"
            else
                @session_info[id] = 0
            end
            return packet
        }

        # server filter
        @proxy.server_filter = lambda { |packet|
            #puts "From server: #{packet.inspect}"
            return packet
        }
        @proxy.run
    end
end

proxy = MQTTProxy.new(
    :local_host => '0.0.0.0',
    :local_port => 5000,
    :server_host => 'test.mosquitto.org',
    :server_port => 1883
)
proxy.run

# name : client.rb
require 'rubygems'
require 'mqtt'
require 'readline'

class MqttClient
    attr_reader :server_host
    attr_reader :server_port
    attr_reader :server
    attr_reader :logger
    attr_reader :client_session
    attr_reader :nickname

    def initialize(args={})
        @server_host = args[:server_host] 
        @server_port = args[:server_port] 
        @nickname = args[:nickname] 
        @server = nil

        @logger = args[:logger]
        if @logger.nil?
            @logger = Logger.new(STDOUT)
            @logger.level = Logger::INFO
        end
    end

    # Login 메서드 
    # 123456을 하드 코딩했다.
    def login session 
        @server.publish "service/auth", session
    end

    def pub topic, message
        @server.publish topic, message 
    end
    def sub topic
        puts @server.get topic
    end

    def run
        begin
            @server = MQTT::Client.connect(
                :host => @server_host, 
                :port => @server_port
            )
        rescue Exception=>  exp
            puts "Error"
        end
        # User Auth 
        login "123456"

        # Publish Thread
        Thread.new do
            while message = Readline.readline("", true)
                pub "/private/123456", "#{@nickname} -> #{message}"
            end
        end

        # Subscribe
        loop do
            sub "/private/123456"
        end
    end
end

nickname = ARGV[0] 

client = MqttClient.new( 
    server_host: "localhost",
    server_port: 5000,
    client_session: "123456",
    nickname: nickname)

client.run

• 클라이언트를 띄우면 server/auth topic로 session 값(123456)을 보내는 것으로 인증을 성공하고, /private/123456 Topic에 subscribe 한다.
• 또 다른 클라이언트를 적당한 nickname으로 띄운 다음 메시지를 전송하면, 최초 클라이언트로 메시지가 push 되는 걸 확인할 수 있다. 
  
  

• MQTT proxy를 이용해서, 유저와 서버의 MQTT 메시지를 연산하는 방식으로 인증, 보안, 필터링 등 Push 서비스를 위한 필요한 기능을 개발할 수 있다.
• Proxy 방식은 개발이 단순하다는 장점이 있다. 반면 성능의 희생이 있을 수 있다. 성능이 중요한 경우 MQTT 브로커를 직접 개발해도 될 것이다. 내가 구현한다면, 우선 프락시 방식으로 개발할 거다. 그 후 굳이 필요하다면 MQTT 브로커를 직접 만드는 방식을 선택할 거다. 
  
  

1. 소켓 줄이기 : 그룹이 소수의 인스턴스에 몰아 넣을 수 있다면 좋겠는데.. 글쎄.. 이건 방법이 떠오르지 않는다.
2. Ping 패킷 주기 줄이기 : 클라이언트 집단이 Ping 패킷을 보내고 있으므로 현재 전체 서버 상황을 알수 있으며, 그중 내가 속한 그룹을 서비스하는 서버들의 상태를 알 수 있다. 한번의 Ping 패킷으로 여러 서버의 상태 정보를 확인할 수 있으므로 ping 요청을 줄일 수 있다. 
   
   

• QoS 0: 메시지는 한번만 전달하며, 전달 여부를 확인하지 않는다. 이 레벨에서는 메시지가 모두에게 전달되거나 혹은 모두에게 전달이 안되거나 하는 식으로 작동한다. Fire and forget(쏘고 잊어버리기) 모드라고 부르기도 한다. 당연하지만 "빠르게 전달"된다.
• QoS 1: 메시지는 반드시 한번 이상 전달된다. 한번 이상이라는 것에 주의해야 한다. 메시지의 핸드셰이킹을 엄밀하게 추적하지 않기 때문에, 메시지가 두 개 이상 전달 될 수 있다. 브로커에서의 중복전송을 막을 수는 없고, 클라이언트에서 중복전송 메시지를 처리해야 한다. 메시지 id 같은 걸 두는 식으로 처리 가능하다. 가장 무난한 모드다.
• QoS 2: 항상 정확히 한번만 전송된다. QoS 1의 중복전송 문제도 없다. 이를 위해서 핸드셰이킹 과정은 엄밀하게 추적되며, 당연히 느리다. 
  
  

• Ruby MQTT
• Facebook 메신저와 MQTT
• 오픈소스 MQTT 브로커 Mosquitto
• MQTT - wikipedia
• MQTT version 3.1.1