mndConsumer.c 6.0 KB
Newer Older
L
Liu Jicong 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#define _DEFAULT_SOURCE
#include "mndConsumer.h"
S
Shengliang Guan 已提交
18
#include "mndAuth.h"
L
Liu Jicong 已提交
19 20 21 22 23 24 25 26 27
#include "mndDb.h"
#include "mndDnode.h"
#include "mndMnode.h"
#include "mndShow.h"
#include "mndStb.h"
#include "mndTopic.h"
#include "mndTrans.h"
#include "mndUser.h"
#include "mndVgroup.h"
L
Liu Jicong 已提交
28
#include "tcompare.h"
L
Liu Jicong 已提交
29 30 31 32 33
#include "tname.h"

#define MND_CONSUMER_VER_NUMBER 1
#define MND_CONSUMER_RESERVE_SIZE 64

L
Liu Jicong 已提交
34 35 36
static int32_t mndConsumerActionInsert(SSdb *pSdb, SMqConsumerObj *pConsumer);
static int32_t mndConsumerActionDelete(SSdb *pSdb, SMqConsumerObj *pConsumer);
static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pConsumer, SMqConsumerObj *pNewConsumer);
S
Shengliang Guan 已提交
37 38 39
static int32_t mndProcessConsumerMetaMsg(SNodeMsg *pMsg);
static int32_t mndGetConsumerMeta(SNodeMsg *pMsg, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveConsumer(SNodeMsg *pMsg, SShowObj *pShow, char *data, int32_t rows);
L
Liu Jicong 已提交
40
static void    mndCancelGetNextConsumer(SMnode *pMnode, void *pIter);
L
Liu Jicong 已提交
41 42 43

int32_t mndInitConsumer(SMnode *pMnode) {
  SSdbTable table = {.sdbType = SDB_CONSUMER,
L
Liu Jicong 已提交
44
                     .keyType = SDB_KEY_INT64,
L
Liu Jicong 已提交
45 46 47 48 49 50 51 52 53 54 55
                     .encodeFp = (SdbEncodeFp)mndConsumerActionEncode,
                     .decodeFp = (SdbDecodeFp)mndConsumerActionDecode,
                     .insertFp = (SdbInsertFp)mndConsumerActionInsert,
                     .updateFp = (SdbUpdateFp)mndConsumerActionUpdate,
                     .deleteFp = (SdbDeleteFp)mndConsumerActionDelete};

  return sdbSetTable(pMnode->pSdb, table);
}

void mndCleanupConsumer(SMnode *pMnode) {}

S
Shengliang Guan 已提交
56
SMqConsumerObj *mndCreateConsumer(int64_t consumerId, const char *cgroup) {
wafwerar's avatar
wafwerar 已提交
57
  SMqConsumerObj *pConsumer = taosMemoryCalloc(1, sizeof(SMqConsumerObj));
L
Liu Jicong 已提交
58 59 60 61
  if (pConsumer == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }
S
Shengliang Guan 已提交
62 63

  pConsumer->recentRemovedTopics = taosArrayInit(1, sizeof(char *));
L
Liu Jicong 已提交
64 65
  pConsumer->epoch = 1;
  pConsumer->consumerId = consumerId;
L
Liu Jicong 已提交
66
  atomic_store_32(&pConsumer->status, MQ_CONSUMER_STATUS__INIT);
L
Liu Jicong 已提交
67 68 69 70 71
  strcpy(pConsumer->cgroup, cgroup);
  taosInitRWLatch(&pConsumer->lock);
  return pConsumer;
}

L
Liu Jicong 已提交
72 73
SSdbRaw *mndConsumerActionEncode(SMqConsumerObj *pConsumer) {
  terrno = TSDB_CODE_OUT_OF_MEMORY;
S
Shengliang Guan 已提交
74 75

  void   *buf = NULL;
L
Liu Jicong 已提交
76
  int32_t tlen = tEncodeSMqConsumerObj(NULL, pConsumer);
L
Liu Jicong 已提交
77 78 79
  int32_t size = sizeof(int32_t) + tlen + MND_CONSUMER_RESERVE_SIZE;

  SSdbRaw *pRaw = sdbAllocRaw(SDB_CONSUMER, MND_CONSUMER_VER_NUMBER, size);
80
  if (pRaw == NULL) goto CM_ENCODE_OVER;
L
Liu Jicong 已提交
81

wafwerar's avatar
wafwerar 已提交
82
  buf = taosMemoryMalloc(tlen);
L
Liu Jicong 已提交
83 84
  if (buf == NULL) goto CM_ENCODE_OVER;

L
Liu Jicong 已提交
85
  void *abuf = buf;
L
Liu Jicong 已提交
86 87
  tEncodeSMqConsumerObj(&abuf, pConsumer);

L
Liu Jicong 已提交
88
  int32_t dataPos = 0;
L
Liu Jicong 已提交
89 90
  SDB_SET_INT32(pRaw, dataPos, tlen, CM_ENCODE_OVER);
  SDB_SET_BINARY(pRaw, dataPos, buf, tlen, CM_ENCODE_OVER);
L
Liu Jicong 已提交
91 92
  SDB_SET_RESERVE(pRaw, dataPos, MND_CONSUMER_RESERVE_SIZE, CM_ENCODE_OVER);
  SDB_SET_DATALEN(pRaw, dataPos, CM_ENCODE_OVER);
L
Liu Jicong 已提交
93

L
Liu Jicong 已提交
94 95
  terrno = TSDB_CODE_SUCCESS;

96
CM_ENCODE_OVER:
wafwerar's avatar
wafwerar 已提交
97
  taosMemoryFreeClear(buf);
98
  if (terrno != 0) {
99
    mError("consumer:%" PRId64 ", failed to encode to raw:%p since %s", pConsumer->consumerId, pRaw, terrstr());
100 101 102 103
    sdbFreeRaw(pRaw);
    return NULL;
  }

104
  mTrace("consumer:%" PRId64 ", encode to raw:%p, row:%p", pConsumer->consumerId, pRaw, pConsumer);
L
Liu Jicong 已提交
105 106 107
  return pRaw;
}

L
Liu Jicong 已提交
108
SSdbRow *mndConsumerActionDecode(SSdbRaw *pRaw) {
109
  terrno = TSDB_CODE_OUT_OF_MEMORY;
S
Shengliang Guan 已提交
110
  void *buf = NULL;
111

L
Liu Jicong 已提交
112
  int8_t sver = 0;
L
Liu Jicong 已提交
113
  if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto CM_DECODE_OVER;
L
Liu Jicong 已提交
114 115 116

  if (sver != MND_CONSUMER_VER_NUMBER) {
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
L
Liu Jicong 已提交
117
    goto CM_DECODE_OVER;
L
Liu Jicong 已提交
118 119
  }

L
Liu Jicong 已提交
120
  SSdbRow *pRow = sdbAllocRow(sizeof(SMqConsumerObj));
L
Liu Jicong 已提交
121
  if (pRow == NULL) goto CM_DECODE_OVER;
122

L
Liu Jicong 已提交
123
  SMqConsumerObj *pConsumer = sdbGetRowObj(pRow);
L
Liu Jicong 已提交
124
  if (pConsumer == NULL) goto CM_DECODE_OVER;
L
Liu Jicong 已提交
125 126

  int32_t dataPos = 0;
L
Liu Jicong 已提交
127
  int32_t len;
L
Liu Jicong 已提交
128
  SDB_GET_INT32(pRaw, dataPos, &len, CM_DECODE_OVER);
wafwerar's avatar
wafwerar 已提交
129
  buf = taosMemoryMalloc(len);
L
Liu Jicong 已提交
130 131 132
  if (buf == NULL) goto CM_DECODE_OVER;
  SDB_GET_BINARY(pRaw, dataPos, buf, len, CM_DECODE_OVER);
  SDB_GET_RESERVE(pRaw, dataPos, MND_CONSUMER_RESERVE_SIZE, CM_DECODE_OVER);
L
Liu Jicong 已提交
133

L
Liu Jicong 已提交
134 135 136
  if (tDecodeSMqConsumerObj(buf, pConsumer) == NULL) {
    goto CM_DECODE_OVER;
  }
L
Liu Jicong 已提交
137 138 139

  terrno = TSDB_CODE_SUCCESS;

L
Liu Jicong 已提交
140
CM_DECODE_OVER:
wafwerar's avatar
wafwerar 已提交
141
  taosMemoryFreeClear(buf);
L
Liu Jicong 已提交
142
  if (terrno != TSDB_CODE_SUCCESS) {
143
    mError("consumer:%" PRId64 ", failed to decode from raw:%p since %s", pConsumer->consumerId, pRaw, terrstr());
wafwerar's avatar
wafwerar 已提交
144
    taosMemoryFreeClear(pRow);
145
    return NULL;
L
Liu Jicong 已提交
146
  }
L
Liu Jicong 已提交
147 148 149 150

  return pRow;
}

L
Liu Jicong 已提交
151
static int32_t mndConsumerActionInsert(SSdb *pSdb, SMqConsumerObj *pConsumer) {
152
  mTrace("consumer:%" PRId64 ", perform insert action", pConsumer->consumerId);
L
Liu Jicong 已提交
153 154 155
  return 0;
}

L
Liu Jicong 已提交
156
static int32_t mndConsumerActionDelete(SSdb *pSdb, SMqConsumerObj *pConsumer) {
157
  mTrace("consumer:%" PRId64 ", perform delete action", pConsumer->consumerId);
L
Liu Jicong 已提交
158 159 160
  return 0;
}

L
Liu Jicong 已提交
161
static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer, SMqConsumerObj *pNewConsumer) {
162
  mTrace("consumer:%" PRId64 ", perform update action", pOldConsumer->consumerId);
L
Liu Jicong 已提交
163 164

  // TODO handle update
L
Liu Jicong 已提交
165 166
  /*taosWLockLatch(&pOldConsumer->lock);*/
  /*taosWUnLockLatch(&pOldConsumer->lock);*/
L
Liu Jicong 已提交
167 168 169 170

  return 0;
}

L
Liu Jicong 已提交
171
SMqConsumerObj *mndAcquireConsumer(SMnode *pMnode, int64_t consumerId) {
L
Liu Jicong 已提交
172 173
  SSdb           *pSdb = pMnode->pSdb;
  SMqConsumerObj *pConsumer = sdbAcquire(pSdb, SDB_CONSUMER, &consumerId);
L
Liu Jicong 已提交
174
  if (pConsumer == NULL) {
175
    terrno = TSDB_CODE_MND_CONSUMER_NOT_EXIST;
L
Liu Jicong 已提交
176 177 178 179
  }
  return pConsumer;
}

L
Liu Jicong 已提交
180
void mndReleaseConsumer(SMnode *pMnode, SMqConsumerObj *pConsumer) {
L
Liu Jicong 已提交
181 182 183
  SSdb *pSdb = pMnode->pSdb;
  sdbRelease(pSdb, pConsumer);
}