[MQTT / Mosca] Mqtt Packet 에 Command Type 추가하기 (for authentication) #3

[MQTT / Mosca] Mqtt Packet 에 Command Type 추가하기 (for authentication) #1

[MQTT / Mosca] Mqtt Packet 에 Command Type 추가하기 (for authentication) #2

하잇!

아 이어서 정리하려니까 넘 힘들다

언젠가는 내가 이걸 다시 보고 공부를 할까?

왜 자꾸 부질 없는 것 같지... 흑흑

그래도 내가 공부한 거 정리하는 거니까 마저 열씨미 해야징 ㅠㅠ

#MQTT Packet Command 추가하기

Authentication 을 위한 Packet 흐름

1. Connect Packet (Device -> Server)

2. Authenticate Packet (Server -> Device)

3. Authenticate Packet (Device -> Server)

4. Connect ACK Packet (Server -> Device)

2단계를 하고 있었다.

저번 글에서는 #4. mqtt-packet (generate) 에서 모듈 열어보고, authenticate type packet 을 generate 하는 코드 추가해주고 테스트까지 해봤다.

그럼 2단계인 Server -> Device 에서

"Server -" 정도 까지 된 거다.

남은 "> Device" 부분을 이 글에서 볼 거고!

이제 Server 측에서 authenticate type packet 을 보내는 걸 했으니!

Device 에서 받아야한다. 

해당 Packet 을 제대로 받으려면, Authenticate type 을 parsing 할 수 있어야겠지?

Device 쪽 코드에서는 mqtt library 를 쓴다.

그러니 이번에는 mqtt library를 열어볼 거야

#5. mqtt (receive)

Device 에서는 mqtt library를 이용해 client 를 생성하고,Server 에 Connect 한다.

그리고 client 에다 관련 event들을 handling 하는 callback function 들을 달아주는데,

우리는 authenticate type 의 packet 에 대한 event 도 받아야 한다!

그러려면 일단 event를 emit 하는 부분을 찾아봐야 하지 않을까?!

(mqtt/lib/client.js)

MqttClient.prototype._handleConnack = function (packet) {
  var rc = packet.returnCode,
    // TODO: move to protocol
    errors = [
      '',
      'Unacceptable protocol version',
      'Identifier rejected',
      'Server unavailable',
      'Bad username or password',
      'Not authorized'
    ];

  clearTimeout(this.connackTimer);

  if (0 === rc) {
    this.reconnecting = false;
    this.emit('connect', packet);
  } else if (0 < rc) {
    this.emit('error',
        new Error('Connection refused: ' + errors[rc]));
  }
};

그래서 evnet를 emit 하는 부분을 찾아보았다!

저거 말고도 많은데, 일단 connect event 를 emit 하는 부분을 찾아봤더니, 저렇게 되어있었다.

그럼 다음 수순은 뭘까?

저 function 을 호출하는 부분을 찾아야겠지!?

(mqtt/lib/client.js)

MqttClient.prototype._handlePacket = function (packet, done) {
  this.emit('packetreceive', packet);

  switch (packet.cmd) {
    case 'publish':
      this._handlePublish(packet, done);
      break;
    case 'authenticate':
      this.emit('authenticate', packet);
      break;
    case 'puback':
    case 'pubrec':
    case 'pubcomp':
    case 'suback':
    case 'unsuback':
      this._handleAck(packet);
      done();
      break;
    case 'pubrel':
      this._handlePubrel(packet, done);
      break;
    case 'connack':
      this._handleConnack(packet);
      done();
      break;
    case 'pingresp':
      this._handlePingresp(packet);
      done();
      break;
    default:
      // do nothing
      // maybe we should do an error handling
      // or just log it
      break;
  }
};

짜잔!

이제 packet type 에 따라 event를 발생시켜주는 부분을 알았으니,

저기다 authenticate event 를 emit 하는 코드를 추가해주면

client 에서 authenticate event 를 handling 할 수 있을 거야!!

그래서 authenticate case 를 추가해줬다.

이럼 끝일까?

젠젠,, ㅠ

Packet Parsing 이 남았다.

기껏 authenticate packet 을 전송하는 거랑, 해당 event 를 처리할 수 있도록 준비까지 다 마쳐놨으니!

제대로 authenticate packet 타입을 처리해줘야 한다.

parsing 을 어디서 하는지 알아보자.

일단 저 _handlePacket function이 packet 이란 parameter 를 받는 것으로 보아,

어딘가에서 packet을 parsing 해서 _handlePacket function 을 호출해주겠지?

(mqtt/lib/client.js) 요 파일 많이많이 본다

function process () {
  var packet = packets.shift(),
    done = completeParse;
  if (packet) {
    that._handlePacket(packet, process);
  } else {
    completeParse = null;
    done();
  }
}

여기다!

근데 packet 을 packets 라는 변수에서 꺼내오는 걸 보니

packet 은 이 function 이 호출되기 전에 이미 처리되어 packets 에 저장되는 것 같다.

그럼 또 이 process function 이 호출되는 곳을 봐야한다.

(mqtt/lib/client.js)

writable._write = function (buf, enc, done) {
  console.log(buf)
  completeParse = done;
  parser.parse(buf);
  process();
};

this.stream.pipe(writable);

요기 있지!!

여길 보면 stream.pipe 란 애가 있는데, 요주의..

아마 나중에 또 등장할 거다. 한 마지막편쯤에..

어쨌든 저 pipe 란 애의 원래 역할은, 

readable stream 에 pipe 로 writable stream 을 연결해주면,

readable stream 이 읽어들인 data buffer 를 writable stream 으로 토스해준다.

그리고 그 buffer 를 처리해주는 부분이 writable._write 로 정의된 저 function 인 것 같다.

자세한 건 document 참조

여튼 그렇게 읽어들인 buffer 를 처리하는 부분에

내가 찾던 키워드가 있다!

parser.parse(buf)...

저 parser 는 또 뭘까 해서 보니

2편에 나왔던 그 아이다

mqtt-packet..

#6. mqtt-packet (parse)

그럼 다시 봐야지 뭐..

(mqtt-packet/parser.js)

Parser.prototype._parsePayload = function () {
  var result = false

  // Do we have a payload? Do we have enough data to complete the payload?
  // PINGs have no payload
  if (this.packet.length === 0 || this._list.length >= this.packet.length) {

    this._pos = 0

    switch (this.packet.cmd) {
      case 'connect':
        this._parseConnect()
        break
      case 'authenticate':
        this._parseAuth()
        break
      case 'connack':
        this._parseConnack()
        break
      case 'publish':
        this._parsePublish()
        break
      case 'puback':
      case 'pubrec':
      case 'pubrel':
      case 'pubcomp':
        this._parseMessageId()
        break
      case 'subscribe':
        this._parseSubscribe()
        break
      case 'suback':
        this._parseSuback()
        break
      case 'unsubscribe':
        this._parseUnsubscribe()
        break
      case 'unsuback':
        this._parseUnsuback()
        break
      case 'pingreq':
      case 'pingresp':
      case 'disconnect':
        // these are empty, nothing to do
        break
      default:
        this._emitError(new Error('not supported'))
    }

    result = true
  }

  return result
}

흐흐 나 코드 따라가는 것 좀 잘 하는 것 같다.

여튼 여기 등장! 빠밤

그럼 여기다 또 언제나 그래왔듯 authenticate case 를 추가해주면 된다.

근데 또 못보던 function 이 은근슬쩍 등장했다.

그렇다. 만들어줘야한다.

진짜 개귀찮다. 내가 왜 이 짓을 시작했지

역시 코딩은 삽질의 연속이다

인정? 어인정~ 동의? 어 보감~ 동휘? 어 보검~~~~~~~~~~~~~~~

(mqtt-packet/parser.js)

Parser.prototype._parseAuth = function() {
  var protocolId // constants id
    , clientId // Client id
    , password // Password
    , username // Username
    , flags = {}
    , packet = this.packet

  // Parse constants id
  protocolId = this._parseString()
  if (protocolId === null)
    return this._emitError(new Error('cannot parse protocol id'))

  if (protocolId != 'MQTT' && protocolId != 'MQIsdp') {

    return this._emitError(new Error('invalid protocol id'))
  }

  packet.protocolId = protocolId

  // Parse constants version number
  if(this._pos >= this._list.length)
    return this._emitError(new Error('packet too short'))

  packet.protocolVersion = this._list.readUInt8(this._pos)

  if(packet.protocolVersion != 3 && packet.protocolVersion != 4) {

    return this._emitError(new Error('invalid protocol version'))
  }

  this._pos++
  if(this._pos >= this._list.length)
    return this._emitError(new Error('packet too short'))

  // Parse connect flags
  flags.username  = (this._list.readUInt8(this._pos) & constants.USERNAME_MASK)
  flags.password  = (this._list.readUInt8(this._pos) & constants.PASSWORD_MASK)

  packet.clean = (this._list.readUInt8(this._pos) & constants.CLEAN_SESSION_MASK) !== 0
  this._pos++

  // Parse keepalive
  packet.keepalive = this._parseNum()
  if(packet.keepalive === -1)
    return this._emitError(new Error('packet too short'))

  // Parse client ID
  clientId = this._parseString()
  if(clientId === null)
    return this._emitError(new Error('packet too short'))
  packet.clientId = clientId

  // Parse username
  if (flags.username) {
    username = this._parseString()
    if(username === null)
      return this._emitError(new Error('cannot parse username'))
    packet.username = username
  }

  // Parse password
  if(flags.password) {
    password = this._parseBuffer()
    if(password === null)
      return this._emitError(new Error('cannot parse username'))
    packet.password = password
  }

  return packet
}

이번에도 역시 _parseConnect 를 참고했다.

참고하되 저번에 만들어둔 Authenticate Generate Function 을 꼭 고려하면서 짜깁기해야한다!

그때 넣었던 데이터들 잘 참고해서 고쳐줘야함!

요까지 해주면 device 에서 authenticate packet 을 받는 것도 매우 잘 된다!

아래는 받는 코드

client.on('authenticate', function(packet) {
  console.log(packet);
});

아 근데 주의할 점이 있다.

혹여나 (가 아니고 대개 그렇겠지만)

Server process 를 돌리는 project directory 와

Device process 를 돌리는 project directory 가 다르다면

각각 node_modules directory 에서 적합한 library 를 수정해줘야 하는데,

지금까지 한 Authenticate Packet (Server -> Device) 단계에서

Server / Device 둘 모두에서 고쳐줘야 하는 부분이 있다.

constants.js !!

authenticate 타입을

Server 에선 추가해줬는데 Device 에서 안 해줬거나

혹은 그 반대의 상황이어도

보내거나 받는 게 제대로 안 된다.

꼭 확인!

ㅠㅠ 다음편엔 3단계로 넘어갈 예정.

정리 넘 힘들당