| /* |
| * 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. |
| */ |
| #include "VersionedCacheableObjectPartList.hpp" |
| |
| #include <geode/CacheableString.hpp> |
| #include <geode/ExceptionTypes.hpp> |
| |
| #include "CacheableToken.hpp" |
| #include "DiskStoreId.hpp" |
| #include "DiskVersionTag.hpp" |
| #include "ThinClientRegion.hpp" |
| namespace apache { |
| namespace geode { |
| namespace client { |
| |
| const uint8_t VersionedCacheableObjectPartList::FLAG_NULL_TAG = 0; |
| const uint8_t VersionedCacheableObjectPartList::FLAG_FULL_TAG = 1; |
| const uint8_t VersionedCacheableObjectPartList::FLAG_TAG_WITH_NEW_ID = 2; |
| const uint8_t VersionedCacheableObjectPartList::FLAG_TAG_WITH_NUMBER_ID = 3; |
| |
| void VersionedCacheableObjectPartList::toData(DataOutput&) const { |
| throw UnsupportedOperationException( |
| "VersionedCacheableObjectPartList::toData not implemented"); |
| } |
| |
| void VersionedCacheableObjectPartList::readObjectPart( |
| int32_t index, DataInput& input, std::shared_ptr<CacheableKey> keyPtr) { |
| auto objType = input.read(); |
| std::shared_ptr<Cacheable> value; |
| m_byteArray[index] = objType; |
| bool isException = (objType == 2 ? 1 : 0); |
| |
| if (isException) { // Exception case |
| // Skip the exception that is in java serialized format, we cant read it. |
| input.advanceCursor(input.readArrayLength()); |
| const auto exMsg = input.readString(); ////4.1 |
| |
| std::shared_ptr<Exception> ex; |
| if (exMsg == "org.apache.geode.security.NotAuthorizedException") { |
| auto message = "Authorization exception at server: " + exMsg; |
| ex = std::make_shared<NotAuthorizedException>(message); |
| } else { |
| auto message = "Exception at remote server: " + exMsg; |
| ex = std::make_shared<CacheServerException>(message); |
| } |
| m_exceptions->emplace(keyPtr, ex); |
| } else if (m_serializeValues) { |
| // read length |
| int32_t skipLen = input.readArrayLength(); |
| int8_t* bytes = nullptr; |
| if (skipLen > 0) { |
| // readObject |
| bytes = new int8_t[skipLen]; |
| input.readBytesOnly(bytes, skipLen); |
| } |
| m_values->emplace(keyPtr, CacheableBytes::create( |
| std::vector<int8_t>(bytes, bytes + skipLen))); |
| |
| /* adongre |
| * CID 29377: Resource leak (RESOURCE_LEAK)Calling allocation function |
| * "apache::geode::client::DataInput::readBytes(unsigned char **, int *)" on |
| * "bytes". |
| */ |
| _GEODE_SAFE_DELETE_ARRAY(bytes); |
| |
| } else { |
| // set nullptr to indicate that there is no exception for the key on this |
| // index |
| // readObject |
| input.readObject(value); |
| if (m_values) m_values->emplace(keyPtr, value); |
| } |
| } |
| |
| void VersionedCacheableObjectPartList::fromData(DataInput& input) { |
| std::lock_guard<decltype(m_responseLock)> guard(m_responseLock); |
| LOGDEBUG("VersionedCacheableObjectPartList::fromData"); |
| uint8_t flags = input.read(); |
| m_hasKeys = (flags & 0x01) == 0x01; |
| bool hasObjects = (flags & 0x02) == 0x02; |
| m_hasTags = (flags & 0x04) == 0x04; |
| m_regionIsVersioned = (flags & 0x08) == 0x08; |
| m_serializeValues = (flags & 0x10) == 0x10; |
| bool persistent = (flags & 0x20) == 0x20; |
| std::shared_ptr<CacheableString> exMsgPtr; |
| int32_t len = 0; |
| bool valuesNULL = false; |
| int32_t keysOffset = (m_keysOffset != nullptr ? *m_keysOffset : 0); |
| // bool readObjLen = false; |
| // int32_t lenOfObjects = 0; |
| if (m_values == nullptr) { |
| m_values = std::make_shared<HashMapOfCacheable>(); |
| valuesNULL = true; |
| } |
| |
| if (!m_hasKeys && !hasObjects && !m_hasTags) { |
| LOGDEBUG( |
| "VersionedCacheableObjectPartList::fromData: Looks like message has no " |
| "data. Returning,"); |
| } |
| |
| auto localKeys = |
| std::make_shared<std::vector<std::shared_ptr<CacheableKey>>>(); |
| if (m_hasKeys) { |
| len = static_cast<int32_t>(input.readUnsignedVL()); |
| |
| for (int32_t index = 0; index < len; ++index) { |
| auto key = std::dynamic_pointer_cast<CacheableKey>(input.readObject()); |
| if (m_resultKeys != nullptr) { |
| m_resultKeys->push_back(key); |
| } |
| m_tempKeys->push_back(key); |
| localKeys->push_back(key); |
| } |
| } else if (m_keys != nullptr) { |
| LOGDEBUG("VersionedCacheableObjectPartList::fromData: m_keys NOT nullptr"); |
| /* |
| if (m_hasKeys) { |
| int64_t tempLen; |
| input.readUnsignedVL(&tempLen); |
| len = (int32_t)tempLen; |
| }else{ |
| len = m_keys->size(); |
| } |
| lenOfObjects = len; |
| readObjLen = true; |
| for (int32_t index = keysOffset; index < keysOffset + len; ++index) { |
| key = m_keys->at(index); |
| if (m_resultKeys != nullptr) { |
| m_resultKeys->push_back(key); |
| } |
| }*/ |
| } else if (hasObjects) { |
| if (m_keys == nullptr && m_resultKeys == nullptr) { |
| LOGERROR( |
| "VersionedCacheableObjectPartList::fromData: Exception: hasObjects " |
| "is true and m_keys and m_resultKeys are also nullptr"); |
| throw FatalInternalException( |
| "VersionedCacheableObjectPartList: " |
| "hasObjects is true and m_keys is also nullptr"); |
| } else { |
| LOGDEBUG( |
| "VersionedCacheableObjectPartList::fromData m_keys or m_resultKeys " |
| "not null"); |
| } |
| } else { |
| LOGDEBUG( |
| "VersionedCacheableObjectPartList::fromData m_hasKeys, m_keys, " |
| "hasObjects all are nullptr"); |
| } // m_hasKeys else ends here |
| |
| if (hasObjects) { |
| len = static_cast<int32_t>(input.readUnsignedVL()); |
| m_byteArray.resize(len); |
| for (int32_t index = 0; index < len; ++index) { |
| if (m_keys != nullptr && !m_hasKeys) { |
| readObjectPart(index, input, m_keys->at(index + keysOffset)); |
| } else /*if (m_resultKeys != nullptr && m_resultKeys->size() > 0)*/ { |
| readObjectPart(index, input, localKeys->at(index)); |
| } /*else{ |
| LOGERROR("VersionedCacheableObjectPartList::fromData: hasObjects = true |
| but m_keys is nullptr and m_resultKeys== nullptr or m_resultKeys->size=0" |
| ); |
| }*/ |
| } |
| } // hasObjects ends here |
| |
| if (m_hasTags) { |
| len = static_cast<int32_t>(input.readUnsignedVL()); |
| ; |
| m_versionTags.resize(len); |
| std::vector<uint16_t> ids; |
| MemberListForVersionStamp& memberListForVersionStamp = |
| *(m_region->getCacheImpl()->getMemberListForVersionStamp()); |
| for (int32_t index = 0; index < len; index++) { |
| uint8_t entryType = input.read(); |
| std::shared_ptr<VersionTag> versionTag; |
| switch (entryType) { |
| case FLAG_NULL_TAG: { |
| break; |
| } |
| case FLAG_FULL_TAG: { |
| if (persistent) { |
| versionTag = std::shared_ptr<VersionTag>( |
| new DiskVersionTag(memberListForVersionStamp)); |
| } else { |
| versionTag = std::shared_ptr<VersionTag>( |
| new VersionTag(memberListForVersionStamp)); |
| } |
| versionTag->fromData(input); |
| versionTag->replaceNullMemberId(getEndpointMemId()); |
| break; |
| } |
| |
| case FLAG_TAG_WITH_NEW_ID: { |
| if (persistent) { |
| versionTag = std::shared_ptr<VersionTag>( |
| new DiskVersionTag(memberListForVersionStamp)); |
| } else { |
| versionTag = std::shared_ptr<VersionTag>( |
| new VersionTag(memberListForVersionStamp)); |
| } |
| versionTag->fromData(input); |
| ids.push_back(versionTag->getInternalMemID()); |
| break; |
| } |
| |
| case FLAG_TAG_WITH_NUMBER_ID: { |
| if (persistent) { |
| versionTag = std::shared_ptr<VersionTag>( |
| new DiskVersionTag(memberListForVersionStamp)); |
| } else { |
| versionTag = std::shared_ptr<VersionTag>( |
| new VersionTag(memberListForVersionStamp)); |
| } |
| versionTag->fromData(input); |
| auto idNumber = input.readUnsignedVL(); |
| versionTag->setInternalMemID(ids.at(idNumber)); |
| break; |
| } |
| default: { break; } |
| } |
| m_versionTags[index] = versionTag; |
| } |
| } else { // if consistancyEnabled=false, we need to pass empty or |
| // std::shared_ptr<NULL> m_versionTags |
| for (int32_t index = 0; index < len; ++index) { |
| std::shared_ptr<VersionTag> versionTag; |
| m_versionTags[index] = versionTag; |
| } |
| } |
| |
| if (hasObjects) { |
| std::shared_ptr<CacheableKey> key; |
| std::shared_ptr<VersionTag> versionTag; |
| std::shared_ptr<Cacheable> value; |
| |
| for (int32_t index = 0; index < len; ++index) { |
| if (m_keys != nullptr && !m_hasKeys) { |
| key = m_keys->at(index + keysOffset); |
| } else /*if (m_resultKeys != nullptr && m_resultKeys->size() > 0)*/ { |
| key = localKeys->at(index); |
| } /*else{ |
| LOGERROR("VersionedCacheableObjectPartList::fromData: hasObjects = true |
| but m_keys is nullptr AND m_resultKeys=nullptr or m_resultKeys->size=0" |
| ); |
| }*/ |
| |
| const auto& iter = m_values->find(key); |
| value = iter == m_values->end() ? nullptr : iter->second; |
| if (m_byteArray[index] != 3) { // 3 - key not found on server |
| std::shared_ptr<Cacheable> oldValue; |
| if (m_addToLocalCache) { |
| int updateCount = -1; |
| versionTag = m_versionTags[index]; |
| |
| GfErrType err = |
| m_region->putLocal("getAll", false, key, value, oldValue, true, |
| updateCount, m_destroyTracker, versionTag); |
| if (err == GF_CACHE_CONCURRENT_MODIFICATION_EXCEPTION) { |
| LOGDEBUG( |
| "VersionedCacheableObjectPartList::fromData putLocal for key [%s] failed because the cache \ |
| already contains an entry with higher version.", |
| Utils::nullSafeToString(key).c_str()); |
| // replace the value with higher version tag |
| (*m_values)[key] = oldValue; |
| } |
| } // END::m_addToLocalCache |
| else { // m_addToLocalCache = false |
| m_region->getEntry(key, oldValue); |
| // if value has already been received via notification or put by |
| // another thread, then return that |
| if (oldValue != nullptr && !CacheableToken::isInvalid(oldValue)) { |
| // replace the value with new value |
| (*m_values)[key] = oldValue; |
| } |
| } |
| } |
| } |
| } |
| if (m_keysOffset != nullptr) *m_keysOffset += len; |
| if (valuesNULL) m_values = nullptr; |
| } |
| |
| } // namespace client |
| } // namespace geode |
| } // namespace apache |
