| /* |
| * 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. |
| */ |
| |
| #pragma once |
| |
| #ifndef GEODE_TCRMESSAGE_H_ |
| #define GEODE_TCRMESSAGE_H_ |
| |
| #include <atomic> |
| #include <cinttypes> |
| #include <map> |
| #include <string> |
| #include <vector> |
| |
| #include <ace/OS.h> |
| |
| #include <geode/CacheableBuiltins.hpp> |
| #include <geode/CacheableKey.hpp> |
| #include <geode/CacheableString.hpp> |
| #include <geode/CqState.hpp> |
| #include <geode/DataInput.hpp> |
| #include <geode/DataOutput.hpp> |
| #include <geode/ExceptionTypes.hpp> |
| #include <geode/Serializable.hpp> |
| #include <geode/internal/geode_globals.hpp> |
| |
| #include "BucketServerLocation.hpp" |
| #include "EventId.hpp" |
| #include "EventIdMap.hpp" |
| #include "FixedPartitionAttributesImpl.hpp" |
| #include "InterestResultPolicy.hpp" |
| #include "SerializationRegistry.hpp" |
| #include "TcrChunkedContext.hpp" |
| #include "VersionTag.hpp" |
| #include "VersionedCacheableObjectPartList.hpp" |
| |
| namespace apache { |
| namespace geode { |
| namespace client { |
| |
| class TcrMessage; |
| class ThinClientRegion; |
| class ThinClientBaseDM; |
| class TcrMessageHelper; |
| class TcrConnection; |
| class TcrMessagePing; |
| |
| class APACHE_GEODE_EXPORT TcrMessage { |
| private: |
| inline static void writeInt(uint8_t* buffer, uint16_t value); |
| inline static void writeInt(uint8_t* buffer, uint32_t value); |
| inline static void readInt(uint8_t* buffer, uint16_t* value); |
| inline static void readInt(uint8_t* buffer, uint32_t* value); |
| |
| public: |
| typedef enum { |
| /* Server couldn't read message; handle it like a server side |
| exception that needs retries */ |
| NOT_PUBLIC_API_WITH_TIMEOUT = -2, |
| INVALID = -1, |
| REQUEST = 0, |
| RESPONSE /* 1 */, |
| EXCEPTION /* 2 */, |
| REQUEST_DATA_ERROR /* 3 */, |
| DATA_NOT_FOUND_ERROR /* 4 Not in use */, |
| PING /* 5 */, |
| REPLY /* 6 */, |
| PUT /* 7 */, |
| PUT_DATA_ERROR /* 8 */, |
| DESTROY /* 9 */, |
| DESTROY_DATA_ERROR /* 10 */, |
| DESTROY_REGION /* 11 */, |
| DESTROY_REGION_DATA_ERROR /* 12 */, |
| CLIENT_NOTIFICATION /* 13 */, |
| UPDATE_CLIENT_NOTIFICATION /* 14 */, |
| LOCAL_INVALIDATE /* 15 */, |
| LOCAL_DESTROY /* 16 */, |
| LOCAL_DESTROY_REGION /* 17 */, |
| CLOSE_CONNECTION /* 18 */, |
| PROCESS_BATCH /* 19 */, |
| REGISTER_INTEREST /* 20 */, |
| REGISTER_INTEREST_DATA_ERROR /* 21 */, |
| UNREGISTER_INTEREST /* 22 */, |
| UNREGISTER_INTEREST_DATA_ERROR /* 23 */, |
| REGISTER_INTEREST_LIST /* 24 */, |
| UNREGISTER_INTEREST_LIST /* 25 */, |
| UNKNOWN_MESSAGE_TYPE_ERROR /* 26 */, |
| LOCAL_CREATE /* 27 */, |
| LOCAL_UPDATE /* 28 */, |
| CREATE_REGION /* 29 */, |
| CREATE_REGION_DATA_ERROR /* 30 */, |
| MAKE_PRIMARY /* 31 */, |
| RESPONSE_FROM_PRIMARY /* 32 */, |
| RESPONSE_FROM_SECONDARY /* 33 */, |
| QUERY /* 34 */, |
| QUERY_DATA_ERROR /* 35 */, |
| CLEAR_REGION /* 36 */, |
| CLEAR_REGION_DATA_ERROR /* 37 */, |
| CONTAINS_KEY /* 38 */, |
| CONTAINS_KEY_DATA_ERROR /* 39 */, |
| KEY_SET /* 40 */, |
| KEY_SET_DATA_ERROR /* 41 */, |
| EXECUTECQ_MSG_TYPE /* 42 */, |
| EXECUTECQ_WITH_IR_MSG_TYPE /*43 */, |
| STOPCQ_MSG_TYPE /*44*/, |
| CLOSECQ_MSG_TYPE /*45 */, |
| CLOSECLIENTCQS_MSG_TYPE /*46*/, |
| CQDATAERROR_MSG_TYPE /*47 */, |
| GETCQSTATS_MSG_TYPE /*48 */, |
| MONITORCQ_MSG_TYPE /*49 */, |
| CQ_EXCEPTION_TYPE /*50 */, |
| REGISTER_INSTANTIATORS = 51 /* 51 */, |
| PERIODIC_ACK = 52 /* 52 */, |
| CLIENT_READY /* 53 */, |
| CLIENT_MARKER /* 54 */, |
| INVALIDATE_REGION /* 55 */, |
| PUTALL /* 56 */, |
| GET_ALL_DATA_ERROR = 58 /* 58 */, |
| EXECUTE_REGION_FUNCTION = 59 /* 59 */, |
| EXECUTE_REGION_FUNCTION_RESULT /* 60 */, |
| EXECUTE_REGION_FUNCTION_ERROR /* 61 */, |
| EXECUTE_FUNCTION /* 62 */, |
| EXECUTE_FUNCTION_RESULT /* 63 */, |
| EXECUTE_FUNCTION_ERROR /* 64 */, |
| CLIENT_REGISTER_INTEREST = 65 /* 65 */, |
| CLIENT_UNREGISTER_INTEREST = 66, |
| REGISTER_DATASERIALIZERS = 67, |
| REQUEST_EVENT_VALUE = 68, |
| REQUEST_EVENT_VALUE_ERROR = 69, /*69*/ |
| PUT_DELTA_ERROR = 70, /*70*/ |
| GET_CLIENT_PR_METADATA = 71, /*71*/ |
| RESPONSE_CLIENT_PR_METADATA = 72, /*72*/ |
| GET_CLIENT_PARTITION_ATTRIBUTES = 73, /*73*/ |
| RESPONSE_CLIENT_PARTITION_ATTRIBUTES = 74, /*74*/ |
| GET_CLIENT_PR_METADATA_ERROR = 75, /*75*/ |
| GET_CLIENT_PARTITION_ATTRIBUTES_ERROR = 76, /*76*/ |
| USER_CREDENTIAL_MESSAGE = 77, |
| REMOVE_USER_AUTH = 78, |
| EXECUTE_REGION_FUNCTION_SINGLE_HOP = 79, |
| QUERY_WITH_PARAMETERS = 80, |
| SIZE = 81, |
| SIZE_ERROR = 82, |
| INVALIDATE = 83, |
| INVALIDATE_ERROR = 84, |
| COMMIT = 85, |
| COMMIT_ERROR = 86, |
| ROLLBACK = 87, |
| TX_FAILOVER = 88, |
| GET_ENTRY = 89, |
| TX_SYNCHRONIZATION = 90, |
| GET_FUNCTION_ATTRIBUTES = 91, |
| GET_PDX_TYPE_BY_ID = 92, |
| GET_PDX_ID_FOR_TYPE = 93, |
| ADD_PDX_TYPE = 94, |
| ADD_PDX_ENUM = 96, |
| GET_PDX_ID_FOR_ENUM = 97, |
| GET_PDX_ENUM_BY_ID = 98, |
| SERVER_TO_CLIENT_PING = 99, |
| // GATEWAY_RECEIVER_COMMAND = 99, |
| GET_ALL_70 = 100, |
| TOMBSTONE_OPERATION = 103, |
| GETDURABLECQS_MSG_TYPE = 105, |
| GET_DURABLE_CQS_DATA_ERROR = 106, |
| GET_ALL_WITH_CALLBACK = 107, |
| PUT_ALL_WITH_CALLBACK = 108, |
| REMOVE_ALL = 109 |
| |
| } MsgType; |
| |
| static bool isKeepAlive() { return *m_keepalive > 0; } |
| static bool isUserInitiativeOps(const TcrMessage& msg) { |
| int32_t msgType = msg.getMessageType(); |
| |
| if (!msg.isMetaRegion() && |
| !(msgType == TcrMessage::PING || msgType == TcrMessage::PERIODIC_ACK || |
| msgType == TcrMessage::MAKE_PRIMARY || |
| msgType == TcrMessage::CLOSE_CONNECTION || |
| msgType == TcrMessage::CLIENT_READY || |
| msgType == TcrMessage::INVALID || |
| msgType == TcrMessage::MONITORCQ_MSG_TYPE || |
| msgType == TcrMessage::GETCQSTATS_MSG_TYPE || |
| msgType == TcrMessage::REQUEST_EVENT_VALUE || |
| msgType == TcrMessage::GET_CLIENT_PR_METADATA || |
| msgType == TcrMessage::GET_CLIENT_PARTITION_ATTRIBUTES || |
| msgType == TcrMessage::GET_PDX_ID_FOR_TYPE || |
| msgType == TcrMessage::GET_PDX_TYPE_BY_ID || |
| msgType == TcrMessage::ADD_PDX_TYPE || msgType == TcrMessage::SIZE || |
| msgType == TcrMessage::TX_FAILOVER || |
| msgType == TcrMessage::GET_ENTRY || |
| msgType == TcrMessage::TX_SYNCHRONIZATION || |
| msgType == TcrMessage::GET_FUNCTION_ATTRIBUTES || |
| msgType == TcrMessage::ADD_PDX_ENUM || |
| msgType == TcrMessage::GET_PDX_ENUM_BY_ID || |
| msgType == TcrMessage::GET_PDX_ID_FOR_ENUM || |
| msgType == TcrMessage::COMMIT || msgType == TcrMessage::ROLLBACK)) { |
| return true; |
| } |
| return false; |
| } |
| static std::shared_ptr<VersionTag> readVersionTagPart( |
| DataInput& input, uint16_t endpointMemId, |
| MemberListForVersionStamp& memberListForVersionStamp); |
| |
| /* constructors */ |
| void setData(const char* bytearray, int32_t len, uint16_t memId, |
| const SerializationRegistry& serializationRegistry, |
| MemberListForVersionStamp& memberListForVersionStamp); |
| |
| void startProcessChunk(ACE_Semaphore& finalizeSema); |
| // nullptr chunk means that this is the last chunk |
| void processChunk(const uint8_t* chunk, int32_t chunkLen, |
| uint16_t endpointmemId, |
| const uint8_t isLastChunkAndisSecurityHeader = 0x00); |
| /* For creating a region on the java server */ |
| /* Note through this you can only create a sub region on the cache server */ |
| /* also for creating REGISTER_INTEREST regex request */ |
| // TcrMessage( TcrMessage::MsgType msgType, const std::string& str1, |
| // const std::string& str2, InterestResultPolicy interestPolicy = |
| // InterestResultPolicy::NONE, bool isDurable = false, bool isCachingEnabled |
| // = false, bool receiveValues = true, ThinClientBaseDM *connectionDM = |
| // nullptr); |
| |
| void InitializeGetallMsg( |
| const std::shared_ptr<Serializable>& aCallbackArgument); |
| // for multiuser cache close |
| |
| // Updates the early ack byte of the message to reflect that it is a retry op |
| void updateHeaderForRetry(); |
| |
| inline const std::vector<std::shared_ptr<CacheableKey>>* getKeys() const { |
| return m_keyList; |
| } |
| |
| inline const std::string& getRegex() const { return m_regex; } |
| |
| inline InterestResultPolicy getInterestResultPolicy() const { |
| if (m_interestPolicy == 2) { |
| return InterestResultPolicy::KEYS_VALUES; |
| } else if (m_interestPolicy == 1) { |
| return InterestResultPolicy::KEYS; |
| } else { |
| return InterestResultPolicy::NONE; |
| } |
| } |
| |
| Pool* getPool() const; |
| |
| /** |
| * Whether the request is meant to be |
| * sent to PR primary node for single hop. |
| */ |
| inline bool forPrimary() const { |
| return m_msgType == TcrMessage::PUT || m_msgType == TcrMessage::DESTROY || |
| m_msgType == TcrMessage::EXECUTE_REGION_FUNCTION; |
| } |
| |
| inline void initCqMap() { m_cqs = new std::map<std::string, int>(); } |
| |
| inline bool forSingleHop() const { |
| return m_msgType == TcrMessage::PUT || m_msgType == TcrMessage::DESTROY || |
| m_msgType == TcrMessage::REQUEST || |
| m_msgType == TcrMessage::GET_ALL_70 || |
| m_msgType == TcrMessage::GET_ALL_WITH_CALLBACK || |
| m_msgType == TcrMessage::EXECUTE_REGION_FUNCTION || |
| m_msgType == TcrMessage::PUTALL || |
| m_msgType == TcrMessage::PUT_ALL_WITH_CALLBACK; |
| } |
| |
| inline bool forTransaction() const { return m_txId != -1; } |
| |
| /* |
| inline void getSingleHopFlags(bool& forSingleHop, bool& forPrimary) const |
| { |
| if (m_msgType == TcrMessage::PUT || |
| m_msgType == TcrMessage::DESTROY || |
| m_msgType == TcrMessage::REQUEST) { |
| |
| forSingleHop = true; |
| |
| if (m_msgType == TcrMessage::REQUEST) { |
| forPrimary = false; |
| } else { |
| forPrimary = true; |
| } |
| |
| } else { |
| |
| forSingleHop = false; |
| forPrimary = false; |
| |
| } |
| } |
| */ |
| |
| /* destroy the connection */ |
| virtual ~TcrMessage(); |
| |
| const std::string& getRegionName() const; |
| Region* getRegion() const; |
| int32_t getMessageType() const; |
| void setMessageType(int32_t msgType); |
| void setMessageTypeRequest(int32_t msgType); // the msgType of the request |
| // that was made if this is a |
| // reply |
| int32_t getMessageTypeRequest() const; |
| std::shared_ptr<CacheableKey> getKey() const; |
| const std::shared_ptr<CacheableKey>& getKeyRef() const; |
| std::shared_ptr<Cacheable> getValue() const; |
| const std::shared_ptr<Cacheable>& getValueRef() const; |
| std::shared_ptr<Cacheable> getCallbackArgument() const; |
| const std::shared_ptr<Cacheable>& getCallbackArgumentRef() const; |
| |
| const std::map<std::string, int>* getCqs() const; |
| bool getBoolValue() const { return m_boolValue; }; |
| inline const char* getException() { |
| exceptionMessage = Utils::nullSafeToString(m_value); |
| return exceptionMessage.c_str(); |
| } |
| |
| const char* getMsgData() const; |
| const char* getMsgHeader() const; |
| const char* getMsgBody() const; |
| size_t getMsgLength() const; |
| size_t getMsgBodyLength() const; |
| std::shared_ptr<EventId> getEventId() const; |
| |
| int32_t getTransId() const; |
| void setTransId(int32_t txId); |
| |
| std::chrono::milliseconds getTimeout() const; |
| void setTimeout(std::chrono::milliseconds timeout); |
| |
| /* we need a static method to generate ping */ |
| /* The caller should not delete the message since it is global. */ |
| static TcrMessagePing* getPingMessage(CacheImpl* cacheImpl); |
| static TcrMessage* getAllEPDisMess(); |
| /* we need a static method to generate close connection message */ |
| /* The caller should not delete the message since it is global. */ |
| static TcrMessage* getCloseConnMessage(CacheImpl* cacheImpl); |
| static void setKeepAlive(bool keepalive); |
| bool isDurable() const { return m_isDurable; } |
| bool receiveValues() const { return m_receiveValues; } |
| bool hasCqPart() const { return m_hasCqsPart; } |
| uint32_t getMessageTypeForCq() const { return m_msgTypeForCq; } |
| bool isInterestListPassed() const { return m_isInterestListPassed; } |
| bool shouldIgnore() const { return m_shouldIgnore; } |
| int8_t getMetaDataVersion() const { return m_metaDataVersion; } |
| uint32_t getEntryNotFound() const { return m_entryNotFound; } |
| int8_t getserverGroupVersion() const { return m_serverGroupVersion; } |
| std::vector<int8_t>* getFunctionAttributes() { return m_functionAttributes; } |
| |
| // set the DM for chunked response messages |
| void setDM(ThinClientBaseDM* dm) { m_tcdm = dm; } |
| |
| ThinClientBaseDM* getDM() { return m_tcdm; } |
| // set the chunked response handler |
| void setChunkedResultHandler(TcrChunkedResult* chunkedResult) { |
| this->m_isLastChunkAndisSecurityHeader = 0x0; |
| m_chunkedResult = chunkedResult; |
| } |
| TcrChunkedResult* getChunkedResultHandler() { return m_chunkedResult; } |
| void setVersionedObjectPartList( |
| std::shared_ptr<VersionedCacheableObjectPartList> versionObjPartListptr) { |
| m_versionObjPartListptr = versionObjPartListptr; |
| } |
| |
| std::shared_ptr<VersionedCacheableObjectPartList> |
| getVersionedObjectPartList() { |
| return m_versionObjPartListptr; |
| } |
| |
| DataInput* getDelta() { return m_delta.get(); } |
| |
| // getDeltaBytes( ) is called *only* by CqService, returns a CacheableBytes |
| // that |
| // takes ownership of delta bytes. |
| std::shared_ptr<CacheableBytes> getDeltaBytes() { |
| if (m_deltaBytes == nullptr) { |
| return nullptr; |
| } |
| auto retVal = CacheableBytes::create( |
| std::vector<int8_t>(m_deltaBytes, m_deltaBytes + m_deltaBytesLen)); |
| m_deltaBytes = nullptr; |
| return retVal; |
| } |
| |
| bool hasDelta() { return (m_delta != nullptr); } |
| |
| void addSecurityPart(int64_t connectionId, int64_t unique_id, |
| TcrConnection* conn); |
| |
| void addSecurityPart(int64_t connectionId, TcrConnection* conn); |
| |
| int64_t getConnectionId(TcrConnection* conn); |
| |
| int64_t getUniqueId(TcrConnection* conn); |
| |
| void createUserCredentialMessage(TcrConnection* conn); |
| |
| void readSecureObjectPart(DataInput& input, bool defaultString = false, |
| bool isChunk = false, |
| uint8_t isLastChunkWithSecurity = 0); |
| |
| void readUniqueIDObjectPart(DataInput& input); |
| |
| void setMetaRegion(bool isMetaRegion) { m_isMetaRegion = isMetaRegion; } |
| |
| bool isMetaRegion() const { return m_isMetaRegion; } |
| |
| int32_t getNumBuckets() const { return m_bucketCount; } |
| |
| const std::string& getColocatedWith() const { return m_colocatedWith; } |
| |
| const std::string& getPartitionResolver() const { |
| return m_partitionResolverName; |
| } |
| |
| std::vector<std::vector<std::shared_ptr<BucketServerLocation>>>* |
| getMetadata() { |
| return m_metadata; |
| } |
| |
| std::vector<std::shared_ptr<FixedPartitionAttributesImpl>>* getFpaSet() { |
| return m_fpaSet; |
| } |
| |
| std::shared_ptr<CacheableHashSet> getFailedNode() { return m_failedNode; } |
| |
| bool isCallBackArguement() const { return m_isCallBackArguement; } |
| |
| void setCallBackArguement(bool aCallBackArguement) { |
| m_isCallBackArguement = aCallBackArguement; |
| } |
| |
| void setBucketServerLocation( |
| std::shared_ptr<BucketServerLocation> serverLocation) { |
| m_bucketServerLocation = serverLocation; |
| } |
| void setVersionTag(std::shared_ptr<VersionTag> versionTag) { |
| m_versionTag = versionTag; |
| } |
| std::shared_ptr<VersionTag> getVersionTag() { return m_versionTag; } |
| uint8_t hasResult() const { return m_hasResult; } |
| std::shared_ptr<CacheableHashMap> getTombstoneVersions() const { |
| return m_tombstoneVersions; |
| } |
| std::shared_ptr<CacheableHashSet> getTombstoneKeys() const { |
| return m_tombstoneKeys; |
| } |
| |
| bool isFEAnotherHop(); |
| |
| protected: |
| TcrMessage() |
| : m_request(nullptr), |
| m_tcdm(nullptr), |
| m_chunkedResult(nullptr), |
| m_keyList(nullptr), |
| m_region(nullptr), |
| m_timeout(15 /*DEFAULT_TIMEOUT_SECONDS*/), |
| m_metadata(), |
| m_cqs(nullptr), |
| m_messageResponseTimeout(-1), |
| m_delta(nullptr), |
| m_deltaBytes(nullptr), |
| m_fpaSet(), |
| m_functionAttributes(), |
| m_connectionIDBytes(nullptr), |
| m_creds(), |
| m_key(), |
| m_value(nullptr), |
| m_failedNode(), |
| m_callbackArgument(nullptr), |
| m_versionTag(), |
| m_eventid(nullptr), |
| m_vectorPtr(), |
| m_bucketServerLocation(nullptr), |
| m_tombstoneVersions(), |
| m_tombstoneKeys(), |
| m_versionObjPartListptr(), |
| exceptionMessage(), |
| m_regionName("INVALID_REGION_NAME"), |
| m_regex(), |
| m_bucketServerLocations(), |
| m_colocatedWith(), |
| m_partitionResolverName(), |
| m_securityHeaderLength(0), |
| m_msgType(TcrMessage::INVALID), |
| m_msgLength(-1), |
| m_msgTypeRequest(0), |
| m_txId(-1), |
| m_bucketCount(0), |
| m_numCqPart(0), |
| m_msgTypeForCq(0), |
| m_deltaBytesLen(0), |
| m_entryNotFound(0), |
| m_feAnotherHop(false), |
| isSecurityOn(false), |
| m_isLastChunkAndisSecurityHeader(0), |
| m_isSecurityHeaderAdded(false), |
| m_isMetaRegion(false), |
| m_decodeAll(false), |
| m_interestPolicy(0), |
| m_isDurable(false), |
| m_receiveValues(false), |
| m_hasCqsPart(false), |
| m_isInterestListPassed(false), |
| m_shouldIgnore(false), |
| m_metaDataVersion(0), |
| m_serverGroupVersion(0), |
| m_boolValue(0), |
| m_isCallBackArguement(false), |
| m_hasResult(0) {} |
| |
| void handleSpecialFECase(); |
| void writeBytesOnly(const std::shared_ptr<Serializable>& se); |
| std::shared_ptr<Serializable> readCacheableBytes(DataInput& input, |
| int lenObj); |
| std::shared_ptr<Serializable> readCacheableString(DataInput& input, |
| int lenObj); |
| |
| TcrMessage(const TcrMessage&) = delete; |
| TcrMessage& operator=(const TcrMessage&) = delete; |
| |
| // some private methods to handle things internally. |
| void handleByteArrayResponse( |
| const char* bytearray, int32_t len, uint16_t endpointMemId, |
| const SerializationRegistry& serializationRegistry, |
| MemberListForVersionStamp& memberListForVersionStamp); |
| void readObjectPart(DataInput& input, bool defaultString = false); |
| void readFailedNodePart(DataInput& input); |
| void readCallbackObjectPart(DataInput& input, bool defaultString = false); |
| void readKeyPart(DataInput& input); |
| void readBooleanPartAsObject(DataInput& input, bool* boolVal); |
| void readIntPart(DataInput& input, uint32_t* intValue); |
| void readLongPart(DataInput& input, uint64_t* intValue); |
| bool readExceptionPart(DataInput& input, uint8_t isLastChunk, |
| bool skipFirstPart = true); |
| void readVersionTag(DataInput& input, uint16_t endpointMemId, |
| MemberListForVersionStamp& memberListForVersionStamp); |
| void readOldValue(DataInput& input); |
| void readPrMetaData(DataInput& input); |
| void writeObjectPart(const std::shared_ptr<Serializable>& se, |
| bool isDelta = false, bool callToData = false, |
| const std::vector<std::shared_ptr<CacheableKey>>* |
| getAllKeyList = nullptr); |
| void writeHeader(uint32_t msgType, uint32_t numOfParts); |
| void writeRegionPart(const std::string& regionName); |
| void writeStringPart(const std::string& str); |
| void writeEventIdPart(int reserveSize = 0, |
| bool fullValueAfterDeltaFail = false); |
| void writeMessageLength(); |
| void writeInterestResultPolicyPart(InterestResultPolicy policy); |
| void writeIntPart(int32_t intValue); |
| void writeBytePart(uint8_t byteValue); |
| void writeMillisecondsPart(std::chrono::milliseconds millis); |
| void writeByteAndTimeOutPart(uint8_t byteValue, |
| std::chrono::milliseconds timeout); |
| void chunkSecurityHeader(int skipParts, const uint8_t* bytes, int32_t len, |
| uint8_t isLastChunkAndSecurityHeader); |
| |
| void readEventIdPart(DataInput& input, bool skip = false, |
| int32_t parts = 1); // skip num parts then read eventid |
| |
| void skipParts(DataInput& input, int32_t numParts = 1); |
| void readStringPart(DataInput& input, uint32_t* len, char** str); |
| void readCqsPart(DataInput& input); |
| void readHashMapForGCVersions(apache::geode::client::DataInput& input, |
| std::shared_ptr<CacheableHashMap>& value); |
| void readHashSetForGCVersions(apache::geode::client::DataInput& input, |
| std::shared_ptr<CacheableHashSet>& value); |
| std::shared_ptr<DSMemberForVersionStamp> readDSMember( |
| apache::geode::client::DataInput& input); |
| |
| std::unique_ptr<DataOutput> m_request; |
| /** the associated region that is handling processing of chunked responses */ |
| ThinClientBaseDM* m_tcdm; |
| TcrChunkedResult* m_chunkedResult; |
| const std::vector<std::shared_ptr<CacheableKey>>* m_keyList; |
| const Region* m_region; |
| std::chrono::milliseconds m_timeout; |
| std::vector<std::vector<std::shared_ptr<BucketServerLocation>>>* m_metadata; |
| std::map<std::string, int>* m_cqs; |
| std::chrono::milliseconds m_messageResponseTimeout; |
| std::unique_ptr<DataInput> m_delta; |
| int8_t* m_deltaBytes; |
| std::vector<std::shared_ptr<FixedPartitionAttributesImpl>>* m_fpaSet; |
| std::vector<int8_t>* m_functionAttributes; |
| std::shared_ptr<CacheableBytes> m_connectionIDBytes; |
| std::shared_ptr<Properties> m_creds; |
| std::shared_ptr<CacheableKey> m_key; |
| std::shared_ptr<Cacheable> m_value; |
| std::shared_ptr<CacheableHashSet> m_failedNode; |
| std::shared_ptr<Cacheable> m_callbackArgument; |
| std::shared_ptr<VersionTag> m_versionTag; |
| std::shared_ptr<EventId> m_eventid; |
| std::shared_ptr<CacheableVector> m_vectorPtr; |
| std::shared_ptr<BucketServerLocation> m_bucketServerLocation; |
| std::shared_ptr<CacheableHashMap> m_tombstoneVersions; |
| std::shared_ptr<CacheableHashSet> m_tombstoneKeys; |
| std::shared_ptr<VersionedCacheableObjectPartList> m_versionObjPartListptr; |
| std::string exceptionMessage; |
| std::string m_regionName; |
| std::string m_regex; |
| std::vector<std::shared_ptr<BucketServerLocation>> m_bucketServerLocations; |
| std::string m_colocatedWith; |
| std::string m_partitionResolverName; |
| int32_t m_securityHeaderLength; |
| int32_t m_msgType; |
| int32_t m_msgLength; |
| /** the msgType of the request if this TcrMessage is a reply msg */ |
| int32_t m_msgTypeRequest; |
| int32_t m_txId; |
| int32_t m_bucketCount; |
| uint32_t m_numCqPart; |
| uint32_t m_msgTypeForCq; |
| int32_t m_deltaBytesLen; |
| uint32_t m_entryNotFound; |
| bool m_feAnotherHop; |
| bool isSecurityOn; |
| uint8_t m_isLastChunkAndisSecurityHeader; |
| bool m_isSecurityHeaderAdded; |
| bool m_isMetaRegion; |
| /** used only when decoding reply message, if false, decode header only */ |
| bool m_decodeAll; |
| char m_interestPolicy; |
| bool m_isDurable; |
| bool m_receiveValues; |
| bool m_hasCqsPart; |
| bool m_isInterestListPassed; |
| bool m_shouldIgnore; |
| int8_t m_metaDataVersion; |
| int8_t m_serverGroupVersion; |
| bool m_boolValue; |
| bool m_isCallBackArguement; |
| uint8_t m_hasResult; |
| |
| static std::atomic<int32_t> m_transactionId; |
| static uint8_t* m_keepalive; |
| const static int m_flag_empty; |
| const static int m_flag_concurrency_checks; |
| |
| friend class TcrMessageHelper; |
| }; |
| |
| class TcrMessageDestroyRegion : public TcrMessage { |
| public: |
| TcrMessageDestroyRegion( |
| DataOutput* dataOutput, const Region* region, |
| const std::shared_ptr<Serializable>& aCallbackArgument, |
| std::chrono::milliseconds messageResponsetimeout, |
| ThinClientBaseDM* connectionDM); |
| |
| virtual ~TcrMessageDestroyRegion() {} |
| |
| private: |
| }; |
| |
| class TcrMessageClearRegion : public TcrMessage { |
| public: |
| TcrMessageClearRegion(DataOutput* dataOutput, const Region* region, |
| const std::shared_ptr<Serializable>& aCallbackArgument, |
| std::chrono::milliseconds messageResponsetimeout, |
| ThinClientBaseDM* connectionDM); |
| |
| virtual ~TcrMessageClearRegion() {} |
| |
| private: |
| }; |
| |
| class TcrMessageQuery : public TcrMessage { |
| public: |
| TcrMessageQuery(DataOutput* dataOutput, const std::string& regionName, |
| std::chrono::milliseconds messageResponsetimeout, |
| ThinClientBaseDM* connectionDM); |
| |
| virtual ~TcrMessageQuery() {} |
| |
| private: |
| }; |
| |
| class TcrMessageStopCQ : public TcrMessage { |
| public: |
| TcrMessageStopCQ(DataOutput* dataOutput, const std::string& regionName, |
| std::chrono::milliseconds messageResponsetimeout, |
| ThinClientBaseDM* connectionDM); |
| |
| virtual ~TcrMessageStopCQ() {} |
| |
| private: |
| }; |
| |
| class TcrMessageCloseCQ : public TcrMessage { |
| public: |
| TcrMessageCloseCQ(DataOutput* dataOutput, const std::string& regionName, |
| std::chrono::milliseconds messageResponsetimeout, |
| ThinClientBaseDM* connectionDM); |
| |
| virtual ~TcrMessageCloseCQ() {} |
| |
| private: |
| }; |
| |
| class TcrMessageQueryWithParameters : public TcrMessage { |
| public: |
| TcrMessageQueryWithParameters( |
| DataOutput* dataOutput, const std::string& regionName, |
| const std::shared_ptr<Serializable>& aCallbackArgument, |
| std::shared_ptr<CacheableVector> paramList, |
| std::chrono::milliseconds messageResponsetimeout, |
| ThinClientBaseDM* connectionDM); |
| |
| virtual ~TcrMessageQueryWithParameters() {} |
| |
| private: |
| }; |
| |
| class TcrMessageContainsKey : public TcrMessage { |
| public: |
| TcrMessageContainsKey(DataOutput* dataOutput, const Region* region, |
| const std::shared_ptr<CacheableKey>& key, |
| const std::shared_ptr<Serializable>& aCallbackArgument, |
| bool isContainsKey, ThinClientBaseDM* connectionDM); |
| |
| virtual ~TcrMessageContainsKey() {} |
| |
| private: |
| }; |
| |
| class TcrMessageGetDurableCqs : public TcrMessage { |
| public: |
| TcrMessageGetDurableCqs(DataOutput* dataOutput, |
| ThinClientBaseDM* connectionDM); |
| |
| virtual ~TcrMessageGetDurableCqs() {} |
| |
| private: |
| }; |
| |
| class TcrMessageRequest : public TcrMessage { |
| public: |
| TcrMessageRequest(DataOutput* dataOutput, const Region* region, |
| const std::shared_ptr<CacheableKey>& key, |
| const std::shared_ptr<Serializable>& aCallbackArgument, |
| ThinClientBaseDM* connectionDM = nullptr); |
| |
| virtual ~TcrMessageRequest() {} |
| |
| private: |
| }; |
| |
| class TcrMessageInvalidate : public TcrMessage { |
| public: |
| TcrMessageInvalidate(DataOutput* dataOutput, const Region* region, |
| const std::shared_ptr<CacheableKey>& key, |
| const std::shared_ptr<Serializable>& aCallbackArgument, |
| ThinClientBaseDM* connectionDM = nullptr); |
| |
| private: |
| }; |
| |
| class TcrMessageDestroy : public TcrMessage { |
| public: |
| TcrMessageDestroy(DataOutput* dataOutput, const Region* region, |
| const std::shared_ptr<CacheableKey>& key, |
| const std::shared_ptr<Cacheable>& value, |
| const std::shared_ptr<Serializable>& aCallbackArgument, |
| ThinClientBaseDM* connectionDM = nullptr); |
| |
| private: |
| }; |
| |
| class TcrMessageRegisterInterestList : public TcrMessage { |
| public: |
| TcrMessageRegisterInterestList( |
| DataOutput* dataOutput, const Region* region, |
| const std::vector<std::shared_ptr<CacheableKey>>& keys, |
| bool isDurable = false, bool isCachingEnabled = false, |
| bool receiveValues = true, |
| InterestResultPolicy interestPolicy = InterestResultPolicy::NONE, |
| ThinClientBaseDM* connectionDM = nullptr); |
| |
| virtual ~TcrMessageRegisterInterestList() {} |
| |
| private: |
| }; |
| |
| class TcrMessageUnregisterInterestList : public TcrMessage { |
| public: |
| TcrMessageUnregisterInterestList( |
| DataOutput* dataOutput, const Region* region, |
| const std::vector<std::shared_ptr<CacheableKey>>& keys, |
| bool isDurable = false, bool receiveValues = true, |
| InterestResultPolicy interestPolicy = InterestResultPolicy::NONE, |
| ThinClientBaseDM* connectionDM = nullptr); |
| |
| virtual ~TcrMessageUnregisterInterestList() {} |
| |
| private: |
| }; |
| |
| class TcrMessagePut : public TcrMessage { |
| public: |
| TcrMessagePut(DataOutput* dataOutput, const Region* region, |
| const std::shared_ptr<CacheableKey>& key, |
| const std::shared_ptr<Cacheable>& value, |
| const std::shared_ptr<Serializable>& aCallbackArgument, |
| bool isDelta = false, ThinClientBaseDM* connectionDM = nullptr, |
| bool isMetaRegion = false, bool fullValueAfterDeltaFail = false, |
| const char* regionName = nullptr); |
| |
| virtual ~TcrMessagePut() {} |
| |
| private: |
| }; |
| |
| class TcrMessageCreateRegion : public TcrMessage { |
| public: |
| TcrMessageCreateRegion(DataOutput* dataOutput, const std::string& str1, |
| const std::string& str2, bool isDurable = false, |
| bool receiveValues = true, |
| ThinClientBaseDM* connectionDM = nullptr); |
| |
| virtual ~TcrMessageCreateRegion() {} |
| |
| private: |
| }; |
| |
| class TcrMessageRegisterInterest : public TcrMessage { |
| public: |
| TcrMessageRegisterInterest( |
| DataOutput* dataOutput, const std::string& str1, const std::string& str2, |
| InterestResultPolicy interestPolicy = InterestResultPolicy::NONE, |
| bool isDurable = false, bool isCachingEnabled = false, |
| bool receiveValues = true, ThinClientBaseDM* connectionDM = nullptr); |
| |
| virtual ~TcrMessageRegisterInterest() {} |
| |
| private: |
| }; |
| |
| class TcrMessageUnregisterInterest : public TcrMessage { |
| public: |
| TcrMessageUnregisterInterest( |
| DataOutput* dataOutput, const std::string& str1, const std::string& str2, |
| InterestResultPolicy interestPolicy = InterestResultPolicy::NONE, |
| bool isDurable = false, bool receiveValues = true, |
| ThinClientBaseDM* connectionDM = nullptr); |
| |
| virtual ~TcrMessageUnregisterInterest() {} |
| |
| private: |
| }; |
| |
| class TcrMessageTxSynchronization : public TcrMessage { |
| public: |
| TcrMessageTxSynchronization(DataOutput* dataOutput, int ordinal, int txid, |
| int status); |
| |
| virtual ~TcrMessageTxSynchronization() {} |
| |
| private: |
| }; |
| |
| class TcrMessageClientReady : public TcrMessage { |
| public: |
| explicit TcrMessageClientReady(DataOutput* dataOutput); |
| |
| virtual ~TcrMessageClientReady() {} |
| |
| private: |
| }; |
| |
| class TcrMessageCommit : public TcrMessage { |
| public: |
| explicit TcrMessageCommit(DataOutput* dataOutput); |
| |
| virtual ~TcrMessageCommit() {} |
| |
| private: |
| }; |
| |
| class TcrMessageRollback : public TcrMessage { |
| public: |
| explicit TcrMessageRollback(DataOutput* dataOutput); |
| |
| virtual ~TcrMessageRollback() {} |
| |
| private: |
| }; |
| |
| class TcrMessageTxFailover : public TcrMessage { |
| public: |
| explicit TcrMessageTxFailover(DataOutput* dataOutput); |
| |
| virtual ~TcrMessageTxFailover() {} |
| |
| private: |
| }; |
| |
| class TcrMessageMakePrimary : public TcrMessage { |
| public: |
| TcrMessageMakePrimary(DataOutput* dataOutput, bool processedMarker); |
| |
| virtual ~TcrMessageMakePrimary() {} |
| |
| private: |
| }; |
| |
| class TcrMessagePutAll : public TcrMessage { |
| public: |
| TcrMessagePutAll(DataOutput* dataOutput, const Region* region, |
| const HashMapOfCacheable& map, |
| std::chrono::milliseconds messageResponsetimeout, |
| ThinClientBaseDM* connectionDM, |
| const std::shared_ptr<Serializable>& aCallbackArgument); |
| |
| virtual ~TcrMessagePutAll() {} |
| |
| private: |
| }; |
| |
| class TcrMessageRemoveAll : public TcrMessage { |
| public: |
| TcrMessageRemoveAll(DataOutput* dataOutput, const Region* region, |
| const std::vector<std::shared_ptr<CacheableKey>>& keys, |
| const std::shared_ptr<Serializable>& aCallbackArgument, |
| ThinClientBaseDM* connectionDM = nullptr); |
| |
| virtual ~TcrMessageRemoveAll() {} |
| |
| private: |
| }; |
| |
| class TcrMessageExecuteCq : public TcrMessage { |
| public: |
| TcrMessageExecuteCq(DataOutput* dataOutput, const std::string& str1, |
| const std::string& str2, CqState state, bool isDurable, |
| ThinClientBaseDM* connectionDM); |
| |
| virtual ~TcrMessageExecuteCq() {} |
| |
| private: |
| }; |
| |
| class TcrMessageExecuteCqWithIr : public TcrMessage { |
| public: |
| TcrMessageExecuteCqWithIr(DataOutput* dataOutput, const std::string& str1, |
| const std::string& str2, CqState state, |
| bool isDurable, ThinClientBaseDM* connectionDM); |
| |
| virtual ~TcrMessageExecuteCqWithIr() {} |
| |
| private: |
| }; |
| |
| class TcrMessageExecuteRegionFunction : public TcrMessage { |
| public: |
| TcrMessageExecuteRegionFunction( |
| DataOutput* dataOutput, const std::string& funcName, const Region* region, |
| const std::shared_ptr<Cacheable>& args, |
| std::shared_ptr<CacheableVector> routingObj, uint8_t getResult, |
| std::shared_ptr<CacheableHashSet> failedNodes, |
| std::chrono::milliseconds timeout, |
| ThinClientBaseDM* connectionDM = nullptr, int8_t reExecute = 0); |
| |
| virtual ~TcrMessageExecuteRegionFunction() {} |
| |
| private: |
| }; |
| |
| class TcrMessageExecuteRegionFunctionSingleHop : public TcrMessage { |
| public: |
| TcrMessageExecuteRegionFunctionSingleHop( |
| DataOutput* dataOutput, const std::string& funcName, const Region* region, |
| const std::shared_ptr<Cacheable>& args, |
| std::shared_ptr<CacheableHashSet> routingObj, uint8_t getResult, |
| std::shared_ptr<CacheableHashSet> failedNodes, bool allBuckets, |
| std::chrono::milliseconds timeout, ThinClientBaseDM* connectionDM); |
| |
| virtual ~TcrMessageExecuteRegionFunctionSingleHop() {} |
| |
| private: |
| }; |
| |
| class TcrMessageGetClientPartitionAttributes : public TcrMessage { |
| public: |
| TcrMessageGetClientPartitionAttributes(DataOutput* dataOutput, |
| const char* regionName); |
| |
| virtual ~TcrMessageGetClientPartitionAttributes() {} |
| |
| private: |
| }; |
| |
| class TcrMessageGetClientPrMetadata : public TcrMessage { |
| public: |
| TcrMessageGetClientPrMetadata(DataOutput* dataOutput, const char* regionName); |
| |
| virtual ~TcrMessageGetClientPrMetadata() {} |
| |
| private: |
| }; |
| |
| class TcrMessageSize : public TcrMessage { |
| public: |
| TcrMessageSize(DataOutput* dataOutput, const char* regionName); |
| |
| virtual ~TcrMessageSize() {} |
| |
| private: |
| }; |
| |
| class TcrMessageUserCredential : public TcrMessage { |
| public: |
| TcrMessageUserCredential(DataOutput* dataOutput, |
| std::shared_ptr<Properties> creds, |
| ThinClientBaseDM* connectionDM = nullptr); |
| |
| virtual ~TcrMessageUserCredential() {} |
| |
| private: |
| }; |
| |
| class TcrMessageRemoveUserAuth : public TcrMessage { |
| public: |
| TcrMessageRemoveUserAuth(DataOutput* dataOutput, bool keepAlive, |
| ThinClientBaseDM* connectionDM); |
| |
| virtual ~TcrMessageRemoveUserAuth() {} |
| |
| private: |
| }; |
| |
| class TcrMessageGetPdxIdForType : public TcrMessage { |
| public: |
| TcrMessageGetPdxIdForType(DataOutput* dataOutput, |
| const std::shared_ptr<Cacheable>& pdxType, |
| ThinClientBaseDM* connectionDM); |
| |
| virtual ~TcrMessageGetPdxIdForType() {} |
| |
| private: |
| }; |
| |
| class TcrMessageAddPdxType : public TcrMessage { |
| public: |
| TcrMessageAddPdxType(DataOutput* dataOutput, |
| const std::shared_ptr<Cacheable>& pdxType, |
| ThinClientBaseDM* connectionDM, int32_t pdxTypeId = 0); |
| |
| virtual ~TcrMessageAddPdxType() {} |
| |
| private: |
| }; |
| |
| class TcrMessageGetPdxIdForEnum : public TcrMessage { |
| public: |
| TcrMessageGetPdxIdForEnum(DataOutput* dataOutput, |
| const std::shared_ptr<Cacheable>& pdxType, |
| ThinClientBaseDM* connectionDM); |
| |
| virtual ~TcrMessageGetPdxIdForEnum() {} |
| |
| private: |
| }; |
| |
| class TcrMessageAddPdxEnum : public TcrMessage { |
| public: |
| TcrMessageAddPdxEnum(DataOutput* dataOutput, |
| const std::shared_ptr<Cacheable>& pdxType, |
| ThinClientBaseDM* connectionDM, int32_t pdxTypeId = 0); |
| |
| virtual ~TcrMessageAddPdxEnum() {} |
| |
| private: |
| }; |
| |
| class TcrMessageGetPdxTypeById : public TcrMessage { |
| public: |
| TcrMessageGetPdxTypeById(DataOutput* dataOutput, int32_t typeId, |
| ThinClientBaseDM* connectionDM); |
| |
| virtual ~TcrMessageGetPdxTypeById() {} |
| |
| private: |
| }; |
| |
| class TcrMessageGetPdxEnumById : public TcrMessage { |
| public: |
| TcrMessageGetPdxEnumById(DataOutput* dataOutput, int32_t typeId, |
| ThinClientBaseDM* connectionDM); |
| |
| virtual ~TcrMessageGetPdxEnumById() {} |
| |
| private: |
| }; |
| |
| class TcrMessageGetFunctionAttributes : public TcrMessage { |
| public: |
| TcrMessageGetFunctionAttributes(DataOutput* dataOutput, |
| const std::string& funcName, |
| ThinClientBaseDM* connectionDM = nullptr); |
| |
| virtual ~TcrMessageGetFunctionAttributes() {} |
| |
| private: |
| }; |
| |
| class TcrMessageKeySet : public TcrMessage { |
| public: |
| TcrMessageKeySet(DataOutput* dataOutput, const std::string& funcName, |
| ThinClientBaseDM* connectionDM = nullptr); |
| |
| virtual ~TcrMessageKeySet() {} |
| |
| private: |
| }; |
| |
| class TcrMessageRequestEventValue : public TcrMessage { |
| public: |
| TcrMessageRequestEventValue(DataOutput* dataOutput, |
| std::shared_ptr<EventId> eventId); |
| |
| virtual ~TcrMessageRequestEventValue() {} |
| |
| private: |
| }; |
| |
| class TcrMessagePeriodicAck : public TcrMessage { |
| public: |
| TcrMessagePeriodicAck(DataOutput* dataOutput, |
| const EventIdMapEntryList& entries); |
| |
| virtual ~TcrMessagePeriodicAck() {} |
| |
| private: |
| }; |
| |
| class TcrMessageUpdateClientNotification : public TcrMessage { |
| public: |
| TcrMessageUpdateClientNotification(DataOutput* dataOutput, int32_t port); |
| |
| virtual ~TcrMessageUpdateClientNotification() {} |
| |
| private: |
| }; |
| |
| class TcrMessageGetAll : public TcrMessage { |
| public: |
| TcrMessageGetAll( |
| DataOutput* dataOutput, const Region* region, |
| const std::vector<std::shared_ptr<CacheableKey>>* keys, |
| ThinClientBaseDM* connectionDM = nullptr, |
| const std::shared_ptr<Serializable>& aCallbackArgument = nullptr); |
| |
| virtual ~TcrMessageGetAll() {} |
| |
| private: |
| }; |
| |
| class TcrMessageExecuteFunction : public TcrMessage { |
| public: |
| TcrMessageExecuteFunction(DataOutput* dataOutput, const std::string& funcName, |
| const std::shared_ptr<Cacheable>& args, |
| uint8_t getResult, ThinClientBaseDM* connectionDM, |
| std::chrono::milliseconds timeout); |
| |
| virtual ~TcrMessageExecuteFunction() {} |
| |
| private: |
| }; |
| |
| class TcrMessagePing : public TcrMessage { |
| public: |
| TcrMessagePing(DataOutput* dataOutput, bool decodeAll); |
| |
| virtual ~TcrMessagePing() {} |
| |
| private: |
| }; |
| |
| class TcrMessageCloseConnection : public TcrMessage { |
| public: |
| TcrMessageCloseConnection(DataOutput* dataOutput, bool decodeAll); |
| |
| virtual ~TcrMessageCloseConnection() {} |
| |
| private: |
| }; |
| |
| class TcrMessageClientMarker : public TcrMessage { |
| public: |
| TcrMessageClientMarker(DataOutput* dataOutput, bool decodeAll); |
| |
| virtual ~TcrMessageClientMarker() {} |
| |
| private: |
| }; |
| |
| class TcrMessageReply : public TcrMessage { |
| public: |
| TcrMessageReply(bool decodeAll, ThinClientBaseDM* connectionDM); |
| |
| virtual ~TcrMessageReply() {} |
| }; |
| |
| /** |
| * Helper class to invoke some internal methods of TcrMessage. Add any |
| * methods that response processor methods require to access here. |
| */ |
| class TcrMessageHelper { |
| public: |
| /** |
| * result types returned by readChunkPartHeader |
| */ |
| enum class ChunkObjectType { OBJECT = 0, EXCEPTION = 1, NULL_OBJECT = 2 }; |
| |
| /** |
| * Tries to read an exception part and returns true if the exception |
| * was successfully read. |
| */ |
| inline static bool readExceptionPart(TcrMessage& msg, DataInput& input, |
| uint8_t isLastChunk) { |
| return msg.readExceptionPart(input, isLastChunk); |
| } |
| |
| inline static void skipParts(TcrMessage& msg, DataInput& input, |
| int32_t numParts = 1) { |
| msg.skipParts(input, numParts); |
| } |
| |
| /** |
| * Reads header of a chunk part. Returns true if header was successfully |
| * read and false if it is a chunk exception part. |
| * Throws a MessageException with relevant message if an unknown |
| * message type is encountered in the header. |
| */ |
| inline static ChunkObjectType readChunkPartHeader( |
| TcrMessage& msg, DataInput& input, DSCode expectedFirstType, |
| int32_t expectedPartType, const char* methodName, uint32_t& partLen, |
| uint8_t isLastChunk) { |
| partLen = input.readInt32(); |
| const auto isObj = input.readBoolean(); |
| |
| if (partLen == 0) { |
| // special null object is case for scalar query result |
| return ChunkObjectType::NULL_OBJECT; |
| } else if (!isObj) { |
| // otherwise we're currently always expecting an object |
| char exMsg[256]; |
| ACE_OS::snprintf(exMsg, 255, |
| "TcrMessageHelper::readChunkPartHeader: " |
| "%s: part is not object", |
| methodName); |
| LOGDEBUG("%s ", exMsg); |
| // throw MessageException(exMsg); |
| return ChunkObjectType::EXCEPTION; |
| } |
| |
| auto rawByte = input.read(); |
| auto partType = static_cast<DSCode>(rawByte); |
| auto compId = static_cast<int32_t>(partType); |
| |
| // ugly hack to check for exception chunk |
| if (partType == DSCode::JavaSerializable) { |
| input.reset(); |
| if (TcrMessageHelper::readExceptionPart(msg, input, isLastChunk)) { |
| msg.setMessageType(TcrMessage::EXCEPTION); |
| return ChunkObjectType::EXCEPTION; |
| } else { |
| char exMsg[256]; |
| ACE_OS::snprintf( |
| exMsg, 255, |
| "TcrMessageHelper::readChunkPartHeader: %s: cannot handle " |
| "java serializable object from server", |
| methodName); |
| throw MessageException(exMsg); |
| } |
| } else if (partType == DSCode::NullObj) { |
| // special null object is case for scalar query result |
| return ChunkObjectType::NULL_OBJECT; |
| } |
| |
| // TODO enum - wtf? |
| if (expectedFirstType > DSCode::FixedIDDefault) { |
| if (partType != expectedFirstType) { |
| char exMsg[256]; |
| ACE_OS::snprintf(exMsg, 255, |
| "TcrMessageHelper::readChunkPartHeader: " |
| "%s: got unhandled object class = %" PRId8, |
| methodName, static_cast<int8_t>(partType)); |
| throw MessageException(exMsg); |
| } |
| // This is for GETALL |
| if (expectedFirstType == DSCode::FixedIDShort) { |
| compId = input.readInt16(); |
| } // This is for QUERY or REGISTER INTEREST. |
| else if (expectedFirstType == DSCode::FixedIDByte) { |
| compId = input.read(); |
| } |
| } |
| if (compId != expectedPartType) { |
| char exMsg[256]; |
| ACE_OS::snprintf( |
| exMsg, 255, |
| "TcrMessageHelper::readChunkPartHeader: " |
| "%s: got unhandled object type = %d, expected = %d, raw = %d", |
| methodName, compId, expectedPartType, rawByte); |
| throw MessageException(exMsg); |
| } |
| return ChunkObjectType::OBJECT; |
| } |
| |
| inline static ChunkObjectType readChunkPartHeader(TcrMessage& msg, |
| DataInput& input, |
| const char* methodName, |
| uint32_t& partLen, |
| uint8_t isLastChunk) { |
| partLen = input.readInt32(); |
| const auto isObj = input.readBoolean(); |
| |
| if (partLen == 0) { |
| // special null object is case for scalar query result |
| return ChunkObjectType::NULL_OBJECT; |
| } else if (!isObj) { |
| // otherwise we're currently always expecting an object |
| char exMsg[256]; |
| ACE_OS::snprintf(exMsg, 255, |
| "TcrMessageHelper::readChunkPartHeader: " |
| "%s: part is not object", |
| methodName); |
| throw MessageException(exMsg); |
| } |
| |
| const auto partType = static_cast<const DSCode>(input.read()); |
| // ugly hack to check for exception chunk |
| if (partType == DSCode::JavaSerializable) { |
| input.reset(); |
| if (TcrMessageHelper::readExceptionPart(msg, input, isLastChunk)) { |
| msg.setMessageType(TcrMessage::EXCEPTION); |
| return ChunkObjectType::EXCEPTION; |
| } else { |
| char exMsg[256]; |
| ACE_OS::snprintf( |
| exMsg, 255, |
| "TcrMessageHelper::readChunkPartHeader: %s: cannot handle " |
| "java serializable object from server", |
| methodName); |
| throw MessageException(exMsg); |
| } |
| } else if (partType == DSCode::NullObj) { |
| // special null object is case for scalar query result |
| return ChunkObjectType::NULL_OBJECT; |
| } |
| return ChunkObjectType::OBJECT; |
| } |
| }; |
| } // namespace client |
| } // namespace geode |
| } // namespace apache |
| |
| #endif // GEODE_TCRMESSAGE_H_ |
