GraphQL查询语言实践-实现BoltMQ的Console

上篇文章《GraphQL查询语言学习笔记》学习了GraphQL的语法和介绍Golang的库。最近再使用Golang编写一款分布式消息队列BoltMQ，我将GraphQL用到BoltMQ的web管理UI上，这样更进一步了解GraphQL。

GraphQL 设计篇

有了将GraphQL应用到BoltMQ的web UI上的需求，那如何来做呢。我总结了下：

1. 业务梳理
2. GraphQL API设计
3. GraphQL Schema设计
4. GraphQL 第三方库的选择
5. GraphQL 客户端与服务端的实现

GraphQL API设计

API的设计比较重要，需要开发人员充分理解业务，在业务的基础抽象有查询Graph。例如console是BolotMQ的集群管理UI。需求：

• 管理多个BoltMQ集群
• 查询集群的节点信息
• 查询集群的统计信息
• 查询集群的topic信息
• 查询集群的消息信息
• 查询集群的订阅组信息
• 查询集群的消费进度
• 查询集群的在线消费进程
  意思列举了几个功能点，首先将集群这个概念抽象出来，用户在选定集群的情况下才会做下面的查询操作，所有集群就可以作为第一层，然后再往下梳理。graphql就是将业务抽象成图（树）的形式的。

以下是对console的查询操作的API设计，当然你需要理解BoltMQ的一些知识。你可查看BoltMQ了解。

query clusters($name: String, $like: String, $group: String, $msgId: String!) {
  clusters(name: $name) {
    name
    stats {
      producerNums
      consumerNums
      brokerNums
      namesrvNums
      topicNums
      outTotalTodayNums
      outTotalYestNums
      inTotalTodayNums
      inTotalYestNums
    }
    nodes {
      namesrvAddrs
      brokerNodes {
        role
        addr
        version
        desc
        outTps
        inTps
        outTotalTodayNums
        outTotalYestNums
        inTotalTodayNums
        inTotalYestNums
      }
    }
    topics(like: $like) {
      topic
      type
      isSystem
      store {
        brokerName
        queueId
        maxOffset
        minOffset
        lastUpdateTime
      }
      route {
        queues {
          brokerName
          writeQueueNums
          readQueueNums
          perm
          sysFlag
        }
        brokers {
          brokerName
          brokerAddrs {
            brokerId
            addr
          }
        }
      }
      groups
      consumeConn {
        describe
        conns {
          consumeGroup
          clientId
          clientAddr
          language
          version
          consumeTps
          consumeFromWhere
          consumeType
          diff
          messageModel
        }
      }
      consumeProgress(group: $group) {
        consumeGroup
        tps
        diff
        total
        progress {
          brokerOffset
          consumeOffset
          diff
          brokerName
          queueId
        }
      }
    }
  }
  msg(msgId: $msgId) {
    info {
      msgId
      topic
      flag
      body
      queueId
      storeSize
      queueOffset
      sysFlag
      bornTimestamp
      bornHost
      storeTimestamp
      storeHost
      commitLogOffset
      bodyCRC
      reconsumeTimes
      preparedTransactionOffset
      properties {
        key
        val
      }
    }
    tracks {
      code
      type
      consumeGroup
      desc
    }
  }
}

GraphQL Schema 设计

结合GraphQL API设计类型系统Schema，这里的设计偏后端一些，和API是相辅相成的。在API基础上再次确定返回值的类型以及结构的优化。当然API和Schema的设计可以同时做，也可以分开进行。

console的类型系统

# boltmq contole graphql schema
schema {
    query: Query
    mutation: Mutation
}

# The query type, represents all of the entry points into our object graph
type Query {
    clusters(name: String): [Cluster]!
    msg(name: String, msgId: String!): Message
}

# A Cluster from the boltmq server
type Cluster {
    # The name of cluster
    name: String!
    # The stats info of cluster
    stats: ClusterStats!
    # The node info of cluster
    nodes: ClusterNode!
    # The topics of cluster
    topics(like: String): [Topic]!
}

# A ClusterStats info of boltmq cluster
type ClusterStats {
    # The producer nums of cluster
    producerNums: Int!
    # The consumer nums of cluster
    consumerNums: Int!
    # The broker nums of cluster
    brokerNums: Int!
    # The name server nums of cluster
    namesrvNums: Int!
    # The topic nums of cluster
    topicNums: Int!
    # The cluster consumer msg total number today
    outTotalTodayNums: Int!
    # The cluster consumer msg total number yest
    outTotalYestNums: Int!
    # The cluster producer msg total number yest
    inTotalTodayNums: Int!
    # The cluster producer msg total number today
    inTotalYestNums: Int!
}

# A Cluster node info of boltmq cluster
type ClusterNode {
    # The namesrv addr list fo cluster
    namesrvAddrs: [String!]!
    # The broker node list fo cluster
    brokerNodes: [BrokerNode]!
}

# A Boker node info of boltmq cluster
type BrokerNode {
    # The borker role
    role: Int!
    # The borker addr
    addr: String!
    # The borker server version
    version: String!
    # The borker server describe
    desc: String!
    # The borker server current out tps
    outTps: Float!
    # The borker server current in tps
    inTps: Float!
    # The cluster consumer msg total number today
    outTotalTodayNums: Int!
    # The cluster consumer msg total number yest
    outTotalYestNums: Int!
    # The cluster producer msg total number yest
    inTotalTodayNums: Int!
    # The cluster producer msg total number today
    inTotalYestNums: Int!
}

# A topic info of boltmq cluster
type Topic {
    # The topic name
    topic: String!
    # The topic type
    type: Int!
    # The topic type
    isSystem: Boolean!
    # The topic store
    store: TopicStore!
    # The topic route
    route: TopicRoute!
    # The consume group
    groups: [String!]!
    # The consume connection
    consumeConn: ConsumeConn!
    consumeProgress(group: String): [ConsumeProgress]!
}

# topic type
enum TopicType {
    # normal topic
    NORMAL_TOPIC
    # retry topic
    RETRY_TOPIC
    # deadline queue topic
    DLQ_TOPIC
}

# A topic stroe info of boltmq cluster
type TopicStore {
    # The broker name
    brokerName: String!
    # The queue id
    queueId: Int!
    # The max offset
    maxOffset: Int!
    # The min offset
    minOffset: Int!
    # The last update time
    lastUpdateTime: String!
}

# A topic route info of boltmq cluster
type TopicRoute {
    # The route data of queue
    queues: [QueueData]!
    # The route data of broker
    brokers: [BrokerData]!
}

# A queue route data of topic
type QueueData {
    # The broker name
    brokerName: String!
    # The write queue nums
    writeQueueNums: Int!
    # The read queue nums
    readQueueNums: Int!
    # The permissions of topic on broker
    perm: Int!
    # The permissions of topic on broker
    sysFlag: Int!
}

# A broker route data of topic
type BrokerData {
    # The broker name
    brokerName: String!
    # The broker addrs
    brokerAddrs: [BrokerAddr]!
}

# A broker addr of topic route
type BrokerAddr {
    # The broker id
    brokerId: Int!
    # The broker addr
    addr: String!
}

# consume connection
type ConsumeConn {
    # The describe
    describe: String!
    # The connection
    conns: [Connection]!
}

# connection info
type Connection {
    # The consume group name
    consumeGroup: String!
    # The client id
    clientId: String!
    # The client addr
    clientAddr: String!
    # The language
    language: String!
    # The version
    version: String!
    # The consume tps
    consumeTps: Float!
    # The consume from where
    consumeFromWhere: String!
    # The consume type
    consumeType: Int!
    # The message diff total
    diff: Int!
    # The message model
    messageModel: Int!
}

# consume type
enum ConsumeType {
    # actively consume
    CONSUME_ACTIVELY
    # passively consume
    CONSUME_PASSIVELY
}

# message model
enum MessageModel {
    # broadcasting
    BROADCASTING 
    # clustering
    CLUSTERING
}

# consume progress
type ConsumeProgress {
    # The consume group name
    consumeGroup: String!
    # The consume tps
    tps: Float!
    # The consume diff
    diff: Int!
    # The total
    total: Int!
    # The progress data list
    progress: [ConsumeProgressData]!
}

# consume progress data
type ConsumeProgressData {
    # The broker offset
    brokerOffset: Int!
    # The broker offset
    consumeOffset: Int!
    # The consume diff
    diff: Int!
    # The broker name
    brokerName: String!
    # The queue id
    queueId: Int!
}

# message
type Message {
    # The message base info
    info: MessageInfo!
    # The message track list
    tracks: [MessageTrack]!
}

# message info
type MessageInfo {
    # The message id
    msgId: String!
    # The topic name
    topic: String!
    # The message flag
    flag: Int!
    # The message body
    body: String!
    # The queue id
    queueId: Int!
    # The store size
    storeSize: Int!
    # The queue offset
    queueOffset: Int!
    # The message sys flag
    sysFlag: Int!
    # The born timestamp
    bornTimestamp: String!
    # The born host 
    bornHost: String!
    # The store timestamp
    storeTimestamp: String!
    # The store host 
    storeHost: String!
    # The commitlog offset
    commitLogOffset: Int!
    # The message body crc
    bodyCRC: Int!
    # The reconsume times
    reconsumeTimes: Int!
    # The reconsume times
    preparedTransactionOffset: Int!
    # The properties
    properties: [Property!]!
}

# property, replace map
type Property {
    key: String!
    val: String!
}

# message track
type MessageTrack {
    # The track code, 0: success, non-0: failed
    code: Int!
    # track type
    type: Int!
    # consume group name
    consumeGroup: String!
    # error describe
    desc: String!
}

# track type
enum TrackType {
    # subscribed and consumed
    SUBSCRIBEDANDCONSUMED
    # subscribed but filterd
    SUBSCRIBEDBUTFILTERD
    # subscribed but pull
    SUBSCRIBEDBUTPULL
    # subscribed and not consume yet
    SUBSCRIBEDBUTNOTCONSUMEYET
    # unknow exeption
    UNKNOWEXEPTION
    # not subscribed and not consumed
    NOTSUBSCRIBEDANDNOTCONSUMED
    # consume groupId not online
    CONSUMEGROUPIDNOTONLINE
}

# The mutation type, represents all updates we can make to our data
type Mutation {
    create2UpdateTopic(name: String!, topic: TopicInput!): TopicResponse
    deleteTopic(name: String!, topic: String!): TopicResponse
}

# The input object sent when cluster is creating a new topic
input TopicInput {
    # topic
    topic: String!
    # The read queue nums, optional
    readQueueNums: Int!
    # The write queue nums, optional
    writeQueueNums: Int!
    # The order topic, optional
    order: Boolean! 
    # The unit topic, optional
    unit: Boolean!
}

# Represents a topic for a cluster
interface Response {
    code: Int!
    desc: String!
}

type TopicResponse implements Response {
    code: Int!
    desc: String!
}

GraphQL 代码实现

console使用的neelance/graphql-go库，代码查看。使用的缺点和注意事项：

• neelance使用反射实现
• graphql类型匹配严格，并缺少对于的int,int64等基础类型。
• 接口的实现不优雅，我提了issue，等待改进。
• context无法向下传递

封装统一认证

请求header中取得jwtToken进行验证。

type userClaims struct {
    user
    jwt.StandardClaims
}

type authenticator struct {
}

func (auth *authenticator) Chain(w http.ResponseWriter, r *http.Request, ctx *Context) bool {
    // extract jwt
    jwtToken := r.Header.Get("Authorization")

    // parse tokentoken
    token, err := jwt.ParseWithClaims(jwtToken, &userClaims{}, func(token *jwt.Token) (interface{}, error) {
        if _, ok := token.Method.(*jwt.SigningMethodHMAC); !ok {
            return nil, fmt.Errorf("Unexpected signing method")
        }
        return jwtSecret, nil
    })
    if err != nil {
        http.Error(w, "not authorized", http.StatusUnauthorized)
        return false
    }

    claims, ok := token.Claims.(*userClaims)
    if !ok || !token.Valid {
        http.Error(w, "not authorized", http.StatusUnauthorized)
        return false
    }

    ctx.ctx = context.WithValue(r.Context(), userAuthKey, claims.user)
    return true
}

登录后生成token并返回。

type loginHandler struct {
}

func (h *loginHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    decoder := json.NewDecoder(r.Body)
    userParam := struct {
        Username string `json:"username"`
        Password string `json:"password"`
    }{}

    err := decoder.Decode(&userParam)
    if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }
    defer r.Body.Close()

    if userParam.Username != "admin" || userParam.Password != "admin" {
        http.Error(w, "invalid login", http.StatusUnauthorized)
        return
    }

    //generate token
    expire := time.Now().Add(time.Hour * 1).Unix()
    // Create the Claims
    claims := userClaims{
        user: user{
            UserID:   1,
            UserName: userParam.Username,
            IsAdmin:  true,
        },
        StandardClaims: jwt.StandardClaims{
            ExpiresAt: expire,
            Issuer:    "login",
        },
    }

    token := jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
    signedToken, _ := token.SignedString(jwtSecret)

    //output token
    tokenResponse := struct {
        Token string `json:"token"`
    }{signedToken}
    json.NewEncoder(w).Encode(tokenResponse)
}

GraphQL接口接入认证

srv.mux.Handle(pattern, join(&relay.Handler{Schema: schema}, &authenticator{}))

登录接入

srv.mux.Handle(pattern, &loginHandler{})

以上4点就完成认证功能，详细代码查看console-server

还需功能

• dataloader

对比graphql-go库，graphql-go更佳灵活，当抽象程度差一些。根据自己情况自行选择。