/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#ifndef __MQCLIENTAPIIMPL_H__
#define __MQCLIENTAPIIMPL_H__
#include "AsyncCallback.h"
#include "ClientRPCHook.h"
#include "ClientRemotingProcessor.h"
#include "CommandHeader.h"
#include "HeartbeatData.h"
#include "KVTable.h"
#include "LockBatchBody.h"
#include "MQClientException.h"
#include "MQMessageExt.h"
#include "MQProtos.h"
#include "SendResult.h"
#include "SocketUtil.h"
#include "TcpRemotingClient.h"
#include "TopAddressing.h"
#include "TopicConfig.h"
#include "TopicList.h"
#include "TopicRouteData.h"
#include "UtilAll.h"
#include "VirtualEnvUtil.h"

namespace rocketmq {
//<!wrap all API to net ;
//<!************************************************************************
class MQClientAPIImpl {
 public:
  MQClientAPIImpl(const string& mqClientId,
                  ClientRemotingProcessor* clientRemotingProcessor,
                  int pullThreadNum,
                  uint64_t tcpConnectTimeout,
                  uint64_t tcpTransportTryLockTimeout,
                  string unitName);
  virtual ~MQClientAPIImpl();
  void stopAllTcpTransportThread();
  bool writeDataToFile(string filename, string data, bool isSync);
  string fetchNameServerAddr(const string& NSDomain);
  void updateNameServerAddr(const string& addrs);

  void callSignatureBeforeRequest(const string& addr,
                                  RemotingCommand& request,
                                  const SessionCredentials& session_credentials);
  void createTopic(const string& addr,
                   const string& defaultTopic,
                   TopicConfig topicConfig,
                   const SessionCredentials& sessionCredentials);
  void endTransactionOneway(std::string addr,
                            EndTransactionRequestHeader* requestHeader,
                            std::string remark,
                            const SessionCredentials& sessionCredentials);

  SendResult sendMessage(const string& addr,
                         const string& brokerName,
                         const MQMessage& msg,
                         SendMessageRequestHeader* pRequestHeader,
                         int timeoutMillis,
                         int maxRetrySendTimes,
                         int communicationMode,
                         SendCallback* pSendCallback,
                         const SessionCredentials& sessionCredentials);

  PullResult* pullMessage(const string& addr,
                          PullMessageRequestHeader* pRequestHeader,
                          int timeoutMillis,
                          int communicationMode,
                          PullCallback* pullCallback,
                          void* pArg,
                          const SessionCredentials& sessionCredentials);

  void sendHearbeat(const string& addr, HeartbeatData* pHeartbeatData, const SessionCredentials& sessionCredentials);

  void unregisterClient(const string& addr,
                        const string& clientID,
                        const string& producerGroup,
                        const string& consumerGroup,
                        const SessionCredentials& sessionCredentials);

  TopicRouteData* getTopicRouteInfoFromNameServer(const string& topic,
                                                  int timeoutMillis,
                                                  const SessionCredentials& sessionCredentials);

  TopicList* getTopicListFromNameServer(const SessionCredentials& sessionCredentials);

  int wipeWritePermOfBroker(const string& namesrvAddr, const string& brokerName, int timeoutMillis);

  void deleteTopicInBroker(const string& addr, const string& topic, int timeoutMillis);

  void deleteTopicInNameServer(const string& addr, const string& topic, int timeoutMillis);

  void deleteSubscriptionGroup(const string& addr, const string& groupName, int timeoutMillis);

  string getKVConfigByValue(const string& projectNamespace, const string& projectGroup, int timeoutMillis);

  KVTable getKVListByNamespace(const string& projectNamespace, int timeoutMillis);

  void deleteKVConfigByValue(const string& projectNamespace, const string& projectGroup, int timeoutMillis);

  SendResult processSendResponse(const string& brokerName, const MQMessage& msg, RemotingCommand* pResponse);

  PullResult* processPullResponse(RemotingCommand* pResponse);

  int64 getMinOffset(const string& addr,
                     const string& topic,
                     int queueId,
                     int timeoutMillis,
                     const SessionCredentials& sessionCredentials);

  int64 getMaxOffset(const string& addr,
                     const string& topic,
                     int queueId,
                     int timeoutMillis,
                     const SessionCredentials& sessionCredentials);

  int64 searchOffset(const string& addr,
                     const string& topic,
                     int queueId,
                     uint64_t timestamp,
                     int timeoutMillis,
                     const SessionCredentials& sessionCredentials);

  MQMessageExt* viewMessage(const string& addr,
                            int64 phyoffset,
                            int timeoutMillis,
                            const SessionCredentials& sessionCredentials);

  int64 getEarliestMsgStoretime(const string& addr,
                                const string& topic,
                                int queueId,
                                int timeoutMillis,
                                const SessionCredentials& sessionCredentials);

  void getConsumerIdListByGroup(const string& addr,
                                const string& consumerGroup,
                                vector<string>& cids,
                                int timeoutMillis,
                                const SessionCredentials& sessionCredentials);

  int64 queryConsumerOffset(const string& addr,
                            QueryConsumerOffsetRequestHeader* pRequestHeader,
                            int timeoutMillis,
                            const SessionCredentials& sessionCredentials);

  void updateConsumerOffset(const string& addr,
                            UpdateConsumerOffsetRequestHeader* pRequestHeader,
                            int timeoutMillis,
                            const SessionCredentials& sessionCredentials);

  void updateConsumerOffsetOneway(const string& addr,
                                  UpdateConsumerOffsetRequestHeader* pRequestHeader,
                                  int timeoutMillis,
                                  const SessionCredentials& sessionCredentials);

  void consumerSendMessageBack(MQMessageExt& msg,
                               const string& consumerGroup,
                               int delayLevel,
                               int timeoutMillis,
                               const SessionCredentials& sessionCredentials);

  void lockBatchMQ(const string& addr,
                   LockBatchRequestBody* requestBody,
                   vector<MQMessageQueue>& mqs,
                   int timeoutMillis,
                   const SessionCredentials& sessionCredentials);

  void unlockBatchMQ(const string& addr,
                     UnlockBatchRequestBody* requestBody,
                     int timeoutMillis,
                     const SessionCredentials& sessionCredentials);

  void sendMessageAsync(const string& addr,
                        const string& brokerName,
                        const MQMessage& msg,
                        RemotingCommand& request,
                        SendCallback* pSendCallback,
                        int64 timeoutMilliseconds,
                        int maxRetryTimes = 1,
                        int retrySendTimes = 1);

 private:
  SendResult sendMessageSync(const string& addr,
                             const string& brokerName,
                             const MQMessage& msg,
                             RemotingCommand& request,
                             int timeoutMillis);
  /*
  void sendMessageAsync(const string& addr, const string& brokerName,
                        const MQMessage& msg, RemotingCommand& request,
                        SendCallback* pSendCallback, int64 timeoutMilliseconds);
  */
  PullResult* pullMessageSync(const string& addr, RemotingCommand& request, int timeoutMillis);

  void pullMessageAsync(const string& addr,
                        RemotingCommand& request,
                        int timeoutMillis,
                        PullCallback* pullCallback,
                        void* pArg);

 private:
  unique_ptr<TcpRemotingClient> m_pRemotingClient;
  unique_ptr<TopAddressing> m_topAddressing;
  string m_nameSrvAddr;
  bool m_firstFetchNameSrv;
  string m_mqClientId;
};
}  // namespace rocketmq
//<!***************************************************************************
#endif