text:
#include <thread>
#include <chrono>
#include <ace/OS.h>

#include <geode/SystemProperties.hpp>
#include <geode/AuthInitialize.hpp>

#include "TcrEndpoint.hpp"
#include "ThinClientRegion.hpp"
#include "ThinClientPoolHADM.hpp"
#include "StackTrace.hpp"
#include "CacheImpl.hpp"
#include "Utils.hpp"
#include "DistributedSystemImpl.hpp"
#include "util/exception.hpp"

namespace apache {
namespace geode {
namespace client {

const char* TcrEndpoint::NC_Notification = "NC Notification";

TcrEndpoint::TcrEndpoint(const std::string& name, CacheImpl* cacheImpl,
                         ACE_Semaphore& failoverSema,
                         ACE_Semaphore& cleanupSema,
                         ACE_Semaphore& redundancySema, ThinClientBaseDM* DM,
                         bool isMultiUserMode)
    : m_needToConnectInLock(false),
      m_connectLockCond(m_connectLock),
      m_maxConnections(cacheImpl->getDistributedSystem()
                           .getSystemProperties()
                           .connectionPoolSize()),
      m_notifyConnection(0),
      m_notifyReceiver(0),
      m_numRegionListener(0),
      m_isQueueHosted(false),
      m_cacheImpl(cacheImpl),
      m_uniqueId(0),
      m_isAuthenticated(false),
      m_msgSent(false),
      m_pingSent(false),
      m_numberOfTimesFailed(0),
      m_isMultiUserMode(isMultiUserMode),
      m_name(name),
      m_connected(false),
      m_isActiveEndpoint(false),
      m_numRegions(0),
      m_pingTimeouts(0),
      m_notifyCount(0),
      m_failoverSema(failoverSema),
      m_cleanupSema(cleanupSema),
      m_notificationCleanupSema(0),
      m_redundancySema(redundancySema),
      m_dupCount(0),
      m_serverQueueStatus(NON_REDUNDANT_SERVER),
      m_isServerQueueStatusSet(false),
      m_queueSize(0),
      m_baseDM(DM),
      m_noOfConnRefs(0),
      m_distributedMemId(0) {
  /*
  m_name = Utils::convertHostToCanonicalForm(m_name.c_str() );
  */
}

TcrEndpoint::~TcrEndpoint() {
  m_connected = false;
  m_isActiveEndpoint = false;
  closeConnections();
  {
    // force close the notification channel -- see bug #295
    ACE_Guard<ACE_Recursive_Thread_Mutex> guard(m_notifyReceiverLock);
    if (m_numRegionListener > 0) {
      LOGFINE(
          "Connection to %s still has references "
          "to subscription channel while closing",
          m_name.c_str());
      // fail in dev build to track #295 better in regressions
      GF_DEV_ASSERT(m_numRegionListener == 0);

      m_numRegionListener = 0;
      closeNotification();
    }
  }
  while (m_notifyCount > 0) {
    LOGDEBUG("TcrEndpoint::~TcrEndpoint(): reducing notify count at %d",
             m_notifyCount);
    m_notificationCleanupSema.acquire();
    m_notifyCount--;
  }
  LOGFINE("Connection to %s deleted", m_name.c_str());
}

inline bool TcrEndpoint::needtoTakeConnectLock() {
#ifdef __linux
  if (m_cacheImpl->getDistributedSystem()
          .getSystemProperties()
          .connectWaitTimeout() > std::chrono::seconds::zero()) {
    return m_needToConnectInLock;  // once pipe or other socket error will take
                                   // lock to connect.
  }
  return false;  // once pipe or other socket error will take lock to connect.
#else
  return false;
#endif
}

GfErrType TcrEndpoint::createNewConnectionWL(
    TcrConnection*& newConn, bool isClientNotification, bool isSecondary,
    std::chrono::microseconds connectTimeout) {
  LOGFINE("TcrEndpoint::createNewConnectionWL");
  auto connectWaitTimeout = m_cacheImpl->getDistributedSystem()
                                .getSystemProperties()
                                .connectWaitTimeout();
  ACE_Time_Value interval(connectWaitTimeout);
  ACE_Time_Value stopAt(ACE_OS::gettimeofday());
  stopAt += interval;
  bool connCreated = false;

  while (ACE_OS::gettimeofday() < stopAt) {
    int32_t ret = m_connectLock.acquire(&stopAt);

    LOGFINE(
        "TcrEndpoint::createNewConnectionWL ret = %d interval = %ld error =%s",
        ret, interval.get_msec(), ACE_OS::strerror(ACE_OS::last_error()));

    if (ret != -1) {  // got lock
      try {
        LOGFINE("TcrEndpoint::createNewConnectionWL got lock");
        newConn =
            new TcrConnection(m_cacheImpl->tcrConnectionManager(), m_connected);
        newConn->InitTcrConnection(this, m_name.c_str(), m_ports,
                                   isClientNotification, isSecondary,
                                   connectTimeout);

        connCreated = true;  // to break while loop

        m_needToConnectInLock = false;  // no need to take lock

        m_connectLock.release();
        LOGFINE("New Connection Created");
        break;
      } catch (const TimeoutException&) {
        LOGINFO("Timeout1 in handshake with endpoint[%s]", m_name.c_str());
        m_connectLock.release();
        // throw te;
        return GF_CLIENT_WAIT_TIMEOUT_REFRESH_PRMETADATA;
      } catch (std::exception& ex) {
        m_connectLock.release();
        LOGWARN("Failed1 in handshake with endpoint[%s]: %s", m_name.c_str(),
                ex.what());
        // throw ex;
        return GF_CLIENT_WAIT_TIMEOUT_REFRESH_PRMETADATA;
      } catch (...) {
        LOGWARN("Unknown1 failure in handshake with endpoint[%s]",
                m_name.c_str());
        m_connectLock.release();
        // throw;
        return GF_CLIENT_WAIT_TIMEOUT_REFRESH_PRMETADATA;
      }
    }
  }

  if (!connCreated) {
    LOGFINE("TcrEndpoint::createNewConnectionWL timeout");
    // throwException(TimeoutException("Thread is hanged in connect call"));
    return GF_CLIENT_WAIT_TIMEOUT;
  }

  return GF_NOERR;
}

GfErrType TcrEndpoint::createNewConnection(
    TcrConnection*& newConn, bool isClientNotification, bool isSecondary,
    std::chrono::microseconds connectTimeout, int32_t timeoutRetries,
    bool sendUpdateNotification, bool appThreadRequest) {
  LOGFINE(
      "TcrEndpoint::createNewConnection: connectTimeout =%d "
      "m_needToConnectInLock=%d appThreadRequest =%d",
      connectTimeout.count(), m_needToConnectInLock, appThreadRequest);
  GfErrType err = GF_NOERR;
  newConn = nullptr;
  while (timeoutRetries-- >= 0) {
    try {
      if (newConn == nullptr) {
        if (!needtoTakeConnectLock() || !appThreadRequest) {
          newConn = new TcrConnection(m_cacheImpl->tcrConnectionManager(),
                                      m_connected);
          bool authenticate = newConn->InitTcrConnection(
              this, m_name.c_str(), m_ports, isClientNotification, isSecondary,
              connectTimeout);
          if (authenticate) {
            authenticateEndpoint(newConn);
          }
        } else {
          err = createNewConnectionWL(newConn, isClientNotification,
                                      isSecondary, connectTimeout);
          if (err == GF_CLIENT_WAIT_TIMEOUT ||
              err == GF_CLIENT_WAIT_TIMEOUT_REFRESH_PRMETADATA) {
            break;
          }
        }
        // m_connected = true;
      }
      if (!isClientNotification && sendUpdateNotification) {
        bool notificationStarted;
        {
          ACE_Guard<ACE_Recursive_Thread_Mutex> guard(m_notifyReceiverLock);
          notificationStarted = (m_numRegionListener > 0) || m_isQueueHosted;
        }
        if (notificationStarted) {
          LOGFINE("Sending update notification message to endpoint %s",
                  m_name.c_str());
          TcrMessageUpdateClientNotification updateNotificationMsg(
              newConn->getConnectionManager()
                  .getCacheImpl()
                  ->getCache()
                  ->createDataOutput(),
              static_cast<int32_t>(newConn->getPort()));
          newConn->send(updateNotificationMsg.getMsgData(),
                        updateNotificationMsg.getMsgLength());
        }
      }
      err = GF_NOERR;
      break;
    } catch (const TimeoutException&) {
      LOGINFO("Timeout in handshake with endpoint[%s]", m_name.c_str());
      err = GF_TIMOUT;
      m_needToConnectInLock = true;  // while creating the connection
      std::this_thread::sleep_for(std::chrono::milliseconds(50));
    } catch (const GeodeIOException& ex) {
      LOGINFO("IO error[%d] in handshake with endpoint[%s]: %s",
              ACE_OS::last_error(), m_name.c_str(), ex.what());
      err = GF_IOERR;
      m_needToConnectInLock = true;  // while creating the connection
      break;
    } catch (const AuthenticationFailedException& ex) {
      LOGWARN("Authentication failed in handshake with endpoint[%s]: %s",
              m_name.c_str(), ex.what());
      err = GF_AUTHENTICATION_FAILED_EXCEPTION;
      break;
    } catch (const AuthenticationRequiredException& ex) {
      LOGWARN("Authentication required in handshake with endpoint[%s]: %s",
              m_name.c_str(), ex.what());
      err = GF_AUTHENTICATION_REQUIRED_EXCEPTION;
      break;
    } catch (const CacheServerException& ex) {
      LOGWARN("Exception in handshake on server[%s]: %s", m_name.c_str(),
              ex.what());
      err = GF_CACHESERVER_EXCEPTION;
      break;
    } catch (const Exception& ex) {
      LOGWARN("Failed in handshake with endpoint[%s]: %s", m_name.c_str(),
              ex.what());
      err = GF_MSG;
      break;
    } catch (std::exception& ex) {
      LOGWARN("Failed in handshake with endpoint[%s]: %s", m_name.c_str(),
              ex.what());
      err = GF_MSG;
      break;
    } catch (...) {
      LOGWARN("Unknown failure in handshake with endpoint[%s]", m_name.c_str());
      err = GF_MSG;
      break;
    }
  }
  if (err != GF_NOERR && newConn != nullptr) {
    _GEODE_SAFE_DELETE(newConn);
  }
  return err;
}

void TcrEndpoint::authenticateEndpoint(TcrConnection*& conn) {
  LOGDEBUG(
      "TcrEndpoint::authenticateEndpoint m_isAuthenticated  = %d "
      "this->m_baseDM = %d",
      m_isAuthenticated, m_baseDM);
  if (!m_isAuthenticated && m_baseDM) {
    this->setConnected();
    ACE_Guard<ACE_Recursive_Thread_Mutex> guard(m_endpointAuthenticationLock);
    GfErrType err = GF_NOERR;
    auto creds = getCredentials();

    if (creds != nullptr) {
      LOGDEBUG("TcrEndpoint::authenticateEndpoint got creds from app = %d",
               creds->getSize());
    } else {
      LOGDEBUG("TcrEndpoint::authenticateEndpoint no creds from app ");
    }

    TcrMessageUserCredential request(
        m_cacheImpl->getCache()->createDataOutput(), creds, m_baseDM);

    LOGDEBUG("request is created");
    TcrMessageReply reply(true, this->m_baseDM);
    // err = this->sendRequestToEP(request, reply, ( *it ).int_id_);
    err = this->sendRequestConnWithRetry(request, reply, conn);
    LOGDEBUG("authenticateEndpoint error = %d", err);
    if (err == GF_NOERR) {
      // put the object into local region
      switch (reply.getMessageType()) {
        case TcrMessage::RESPONSE: {
          // nothing to be done;
          break;
        }
        case TcrMessage::EXCEPTION: {
          err = ThinClientRegion::handleServerException("AuthException",
                                                        reply.getException());
          break;
        }
        default: {
          LOGERROR("Unknown message type %d while sending credentials",
                   reply.getMessageType());
          err = GF_MSG;
          break;
        }
      }
    }
    // throw exception if it is not authenticated
    GfErrTypeToException("TcrEndpoint::authenticateEndpoint", err);

    m_isAuthenticated = true;
  }
}
std::shared_ptr<Properties> TcrEndpoint::getCredentials() {
  const auto& distributedSystem = m_cacheImpl->getDistributedSystem();
  const auto& tmpSecurityProperties =
      distributedSystem.getSystemProperties().getSecurityProperties();

  if (const auto& authInitialize = m_cacheImpl->getAuthInitialize()) {
    LOGFINER(
        "Acquired handle to AuthInitialize plugin, "
        "getting credentials for %s",
        m_name.c_str());
    const auto& tmpAuthIniSecurityProperties =
        authInitialize->getCredentials(tmpSecurityProperties, m_name.c_str());
    LOGFINER("Done getting credentials");
    return tmpAuthIniSecurityProperties;
  }
  return nullptr;
}

ServerQueueStatus TcrEndpoint::getFreshServerQueueStatus(
    int32_t& queueSize, bool addToQueue, TcrConnection*& statusConn) {
  GfErrType err = GF_NOERR;
  TcrConnection* newConn;
  ServerQueueStatus status = NON_REDUNDANT_SERVER;

  err = createNewConnection(newConn, false, false,
                            m_cacheImpl->getDistributedSystem()
                                .getSystemProperties()
                                .connectTimeout());
  if (err == GF_NOERR) {
    status = newConn->getServerQueueStatus(queueSize);

    if (status == REDUNDANT_SERVER || status == PRIMARY_SERVER) {
      if (addToQueue) {
        m_opConnections.put(newConn, true);
      } else {
        statusConn = newConn;
      }
      m_connected = true;
      return status;
    } else {
      //  remove port from ports list (which is sent to server in notification
      //  handshake).
      closeConnection(newConn);
      return status;
    }
  }

  return status;
}

GfErrType TcrEndpoint::registerDM(bool clientNotification, bool isSecondary,
                                  bool isActiveEndpoint,
                                  ThinClientBaseDM* distMgr) {
  // Pre-conditions:
  // 1. If this is a secondary server then clientNotification must be true
  GF_DEV_ASSERT(!isSecondary || clientNotification);

  bool connected = false;
  GfErrType err = GF_NOERR;

  ACE_Guard<ACE_Recursive_Thread_Mutex> guard(m_connectionLock);
  // Three cases here:
  // 1. m_connected is false, m_isActiveEndpoint is false and then
  //    if isActiveEndpoint is true, then create 'max' connections
  // 2. m_connected is false, m_isActiveEndpoint is false and then
  //    if isActiveEndpoint is false, then create just one connection
  //    to ping the server
  // 3. m_connected is true, m_isActiveEndpoint is false (i.e. server was
  //    previously not an active endpoint) then if isSecondary is false then
  //    create 'max-1' connections else do nothing
  m_opConnections.reset();
  if (m_maxConnections <= 0) {
    connected = true;
  } else if (!m_isActiveEndpoint) {
    int maxConnections = 0;
    if (isActiveEndpoint) {
      if (m_connected) {
        maxConnections = m_maxConnections - 1;
      } else {
        maxConnections = m_maxConnections;
      }
    } else if (!m_connected) {
      maxConnections = 1;
    }
    if (maxConnections > 0) {
      LOGINFO("Starting Handshake with %s%s",
              (isSecondary ? "secondary server "
                           : (isActiveEndpoint ? "" : "primary server ")),
              m_name.c_str());
      for (int connNum = 0; connNum < maxConnections; ++connNum) {
        TcrConnection* newConn;
        if ((err = createNewConnection(newConn, false, false,
                                       m_cacheImpl->getDistributedSystem()
                                           .getSystemProperties()
                                           .connectTimeout(),
                                       0, m_connected)) != GF_NOERR) {
          m_connected = false;
          m_isActiveEndpoint = false;
          closeConnections();
          return err;
        }
        m_opConnections.put(newConn, true);
      }
      LOGINFO("Handshake with %s%s success",
              (isSecondary ? "secondary server "
                           : (isActiveEndpoint ? "" : "primary server ")),
              m_name.c_str());
      m_connected = true;
      m_isActiveEndpoint = isActiveEndpoint;
    }
  }

  if (m_connected || connected) {
    if (clientNotification) {
      if (distMgr != nullptr) {
        ACE_Guard<ACE_Recursive_Thread_Mutex> guardDistMgrs(m_distMgrsLock);
        m_distMgrs.push_back(distMgr);
      }
      LOGFINEST(
          "Registering subscription "
          "channel for endpoint %s",
          m_name.c_str());
      // setup notification channel for the first region
      ACE_Guard<ACE_Recursive_Thread_Mutex> guard(m_notifyReceiverLock);
      if (m_numRegionListener == 0) {
        if ((err = createNewConnection(m_notifyConnection, true, isSecondary,
                                       m_cacheImpl->getDistributedSystem()
                                               .getSystemProperties()
                                               .connectTimeout() *
                                           3,
                                       0)) != GF_NOERR) {
          m_connected = false;
          m_isActiveEndpoint = false;
          closeConnections();
          LOGWARN("Failed to start subscription channel for endpoint %s",
                  m_name.c_str());
          return err;
        }
        m_notifyReceiver = new Task<TcrEndpoint>(
            this, &TcrEndpoint::receiveNotification, NC_Notification);
        m_notifyReceiver->start();
      }
      ++m_numRegionListener;
      LOGFINEST("Incremented notification region count for endpoint %s to %d",
                m_name.c_str(), m_numRegionListener);
      m_connected = true;
    }
  }

  // Post-conditions:
  // 1. The endpoint should be marked as active, only if m_connected is true
  // 2. If this is not an active endpoint and it is connected then only one
  //    connection + notify channel
  GF_DEV_ASSERT(!m_isActiveEndpoint || m_connected);
#if GF_DEVEL_ASSERTS == 1
  int numConnections = m_opConnections.size();
  if (!m_isActiveEndpoint && !isActiveEndpoint && m_connected &&
      (numConnections != 1 || m_numRegionListener <= 0 ||
       m_notifyReceiver == nullptr)) {
    LOGWARN(
        "Inactive connected endpoint does not have exactly one "
        "connection. Number of connections: %d, number of region listeners: "
        "%d",
        numConnections, m_numRegionListener);
  }
#endif

  return err;
}

void TcrEndpoint::unregisterDM(bool clientNotification,
                               ThinClientBaseDM* distMgr, bool) {
  if (clientNotification) {
    LOGFINEST(
        "Closing subscription "
        "channel for endpoint %s",
        m_name.c_str());
    // close notification channel if there is no region
    ACE_Guard<ACE_Recursive_Thread_Mutex> guard(m_notifyReceiverLock);
    if (m_numRegionListener > 0 && --m_numRegionListener == 0) {
      closeNotification();
    }
    LOGFINEST("Decremented subscription region count for endpoint %s to %d",
              m_name.c_str(), m_numRegionListener);
    if (distMgr != nullptr) {
      ACE_Guard<ACE_Recursive_Thread_Mutex> guardDistMgrs(m_distMgrsLock);
      m_distMgrs.remove(distMgr);
    }
    LOGFINEST("Done unsubscribe for endpoint %s", m_name.c_str());
  }
}

void TcrEndpoint::pingServer(ThinClientPoolDM* poolDM) {
  LOGDEBUG("Sending ping message to endpoint %s", m_name.c_str());
  if (!m_connected || m_noOfConnRefs == 0) {
    LOGFINER("Skipping ping task for disconnected endpoint %s", m_name.c_str());
    return;
  }

  if (!m_msgSent && !m_pingSent) {
    TcrMessagePing* pingMsg = TcrMessage::getPingMessage(m_cacheImpl);
    TcrMessageReply reply(true, nullptr);
    LOGFINEST("Sending ping message to endpoint %s", m_name.c_str());
    GfErrType error;
    if (poolDM != nullptr) {
      error = poolDM->sendRequestToEP(*pingMsg, reply, this);
    } else {
      error = send(*pingMsg, reply);
    }
    LOGFINEST("Sent ping message to endpoint %s with error code %d%s",
              m_name.c_str(), error, error == GF_NOERR ? " (no error)" : "");
    if (error == GF_NOERR) {
      m_pingSent = true;
    }
    if (error == GF_TIMOUT && m_pingTimeouts < 2) {
      ++m_pingTimeouts;
    } else {
      m_pingTimeouts = 0;
      //  Only call setConnectionStatus if the status has changed (non thread
      //  safe check)
      // This is to avoid blocking the ping thread if notification channel takes
      // a long time to
      // complete causing the server to drop the client in the midst of
      // connection establishment.
      bool connected = (error == GF_NOERR)
                           ? (reply.getMessageType() == TcrMessage::REPLY)
                           : false;
      if (m_connected != connected) {
        setConnectionStatus(connected);
      }
    }
    LOGFINEST("Completed sending ping message to endpoint %s", m_name.c_str());
  } else {
    m_msgSent = false;
    m_pingSent = false;
  }
}

bool TcrEndpoint::checkDupAndAdd(std::shared_ptr<EventId> eventid) {
  return m_cacheImpl->tcrConnectionManager().checkDupAndAdd(eventid);
}

int TcrEndpoint::receiveNotification(volatile bool& isRunning) {
  LOGFINE("Started subscription channel for endpoint %s", m_name.c_str());
  while (isRunning) {
    TcrMessageReply* msg = nullptr;
    try {
      size_t dataLen;
      ConnErrType opErr = CONN_NOERR;
      auto data = m_notifyConnection->receive(&dataLen, &opErr,
                                              std::chrono::seconds(5));

      if (opErr == CONN_IOERR) {
        // Endpoint is disconnected, this exception is expected
        LOGFINER(
            "IO exception while receiving subscription event for endpoint %d",
            opErr);
        if (isRunning) {
          setConnectionStatus(false);
          // close notification channel
          ACE_Guard<ACE_Recursive_Thread_Mutex> guard(m_notifyReceiverLock);
          if (m_numRegionListener > 0) {
            m_numRegionListener = 0;
            closeNotification();
          }
        }
        break;
      }

      if (data) {
        msg = new TcrMessageReply(true, m_baseDM);
        msg->initCqMap();
        msg->setData(data, static_cast<int32_t>(dataLen),
                     this->getDistributedMemberID(),
                     *(m_cacheImpl->getSerializationRegistry()),
                     *(m_cacheImpl->getMemberListForVersionStamp()));
        handleNotificationStats(static_cast<int64_t>(dataLen));
        LOGDEBUG("receive notification %d", msg->getMessageType());

        if (!isRunning) {
          _GEODE_SAFE_DELETE(msg);
          break;
        }

        if (msg->getMessageType() == TcrMessage::SERVER_TO_CLIENT_PING) {
          LOGFINE("Received ping from server subscription channel.");
        }

        // ignore some message types like REGISTER_INSTANTIATORS
        if (msg->shouldIgnore()) {
          _GEODE_SAFE_DELETE(msg);
          continue;
        }

        bool isMarker = (msg->getMessageType() == TcrMessage::CLIENT_MARKER);
        if (!msg->hasCqPart()) {
          if (msg->getMessageType() != TcrMessage::CLIENT_MARKER) {
            const std::string& regionFullPath1 = msg->getRegionName();
            std::shared_ptr<Region> region1;
            m_cacheImpl->getRegion(regionFullPath1.c_str(), region1);
            if (region1 != nullptr &&
                !static_cast<ThinClientRegion*>(region1.get())
                     ->getDistMgr()
                     ->isEndpointAttached(this)) {
              // drop event before even processing the eventid for duplicate
              // checking
              LOGFINER("Endpoint %s dropping event for region %s",
                       m_name.c_str(), regionFullPath1.c_str());
              _GEODE_SAFE_DELETE(msg);
              continue;
            }
          }
        }

        if (!checkDupAndAdd(msg->getEventId())) {
          m_dupCount++;
          if (m_dupCount % 100 == 1) {
            LOGFINE("Dropped %dst duplicate notification message", m_dupCount);
          }
          _GEODE_SAFE_DELETE(msg);
          continue;
        }

        if (isMarker) {
          LOGFINE("Got a marker message on endpont %s", m_name.c_str());
          m_cacheImpl->processMarker();
          processMarker();
          _GEODE_SAFE_DELETE(msg);
        } else {
          if (!msg->hasCqPart())  // || msg->isInterestListPassed())
          {
            const std::string& regionFullPath = msg->getRegionName();
            std::shared_ptr<Region> region;
            m_cacheImpl->getRegion(regionFullPath.c_str(), region);
            if (region != nullptr) {
              static_cast<ThinClientRegion*>(region.get())
                  ->receiveNotification(msg);
            } else {
              LOGWARN(
                  "Notification for region %s that does not exist in "
                  "client cacheImpl.",
                  regionFullPath.c_str());
            }
          } else {
            LOGDEBUG("receive cq notification %d", msg->getMessageType());
            auto queryService = getQueryService();
            if (queryService != nullptr) {
              static_cast<RemoteQueryService*>(queryService.get())
                  ->receiveNotification(msg);
            }
          }
        }
      }
    } catch (const TimeoutException&) {
      // If there is no notification, this exception is expected
      // But this is valid only when *no* data has been received
      // otherwise if data has been read then TcrConnection will throw
      // a GeodeIOException which will cause the channel to close.
      LOGDEBUG(
          "receiveNotification timed out: no data received from "
          "endpoint %s",
          m_name.c_str());
    } catch (const GeodeIOException& e) {
      // Endpoint is disconnected, this exception is expected
      LOGFINER(
          "IO exception while receiving subscription event for endpoint %s: %s",
          m_name.c_str(), e.what());
      if (m_connected) {
        setConnectionStatus(false);
        // close notification channel
        ACE_Guard<ACE_Recursive_Thread_Mutex> guard(m_notifyReceiverLock);
        if (m_numRegionListener > 0) {
          m_numRegionListener = 0;
          closeNotification();
        }
      }
      break;
    } catch (const Exception& ex) {
      _GEODE_SAFE_DELETE(msg);
      LOGERROR(
          "Exception while receiving subscription event for endpoint %s:: %s: "
          "%s",
          m_name.c_str(), ex.getName().c_str(), ex.what());
    } catch (...) {
      _GEODE_SAFE_DELETE(msg);
      LOGERROR(
          "Unexpected exception while "
          "receiving subscription event from endpoint %s",
          m_name.c_str());
    }
  }
  LOGFINE("Ended subscription channel for endpoint %s", m_name.c_str());
  return 0;
}

inline bool TcrEndpoint::compareTransactionIds(int32_t reqTransId,
                                               int32_t replyTransId,
                                               std::string& failReason,
                                               TcrConnection* conn) {
  LOGDEBUG("TcrEndpoint::compareTransactionIds requested id = %d ,replied = %d",
           reqTransId, replyTransId);
  if (replyTransId != reqTransId) {
    LOGERROR(
        "Transaction ids do not match on endpoint %s for "
        "send operation: %d, %d. Possible serialization mismatch",
        m_name.c_str(), reqTransId, replyTransId);
    closeConnection(conn);
    failReason = "mismatch of transaction IDs in operation";
    return false;
  }
  return true;
}

inline bool TcrEndpoint::handleIOException(const std::string& message,
                                           TcrConnection*& conn, bool) {
  int32_t lastError = ACE_OS::last_error();
  if (lastError == ECONNRESET || lastError == EPIPE) {
    _GEODE_SAFE_DELETE(conn);
  } else {
    closeConnection(conn);
  }
  LOGFINE(
      "IO error during send for endpoint %s "
      "[errno: %d: %s]: %s",
      m_name.c_str(), lastError, ACE_OS::strerror(lastError), message.c_str());
  // EAGAIN =11, EWOULDBLOCK = 10035L, EPIPE = 32, ECONNRESET =10054L(An
  // existing connection was forcibly closed by the remote host.)
  if (!(lastError == EAGAIN || lastError == EWOULDBLOCK /*||
        lastError == ECONNRESET */
        /*|| lastError == EPIPE*/)) {
    // break from enclosing loop without retries
    // something wrong try connect in lock
    m_needToConnectInLock = true;
    return false;
  }
  std::this_thread::sleep_for(std::chrono::milliseconds(10));
  return true;
}

GfErrType TcrEndpoint::sendRequestConn(const TcrMessage& request,
                                       TcrMessageReply& reply,
                                       TcrConnection* conn,
                                       std::string& failReason) {
  int32_t type = request.getMessageType();
  GfErrType error = GF_NOERR;

  LOGFINER("Sending request type %d to endpoint [%s] via connection [%p]", type,
           m_name.c_str(), conn);
  // TcrMessage * req = const_cast<TcrMessage *>(&request);
  LOGDEBUG("TcrEndpoint::sendRequestConn  = %d", m_baseDM);
  if (m_baseDM != nullptr) m_baseDM->beforeSendingRequest(request, conn);
  if (((type == TcrMessage::EXECUTE_FUNCTION ||
        type == TcrMessage::EXECUTE_REGION_FUNCTION) &&
       (request.hasResult() & 2))) {
    sendRequestForChunkedResponse(request, reply, conn);
  } else if (type == TcrMessage::REGISTER_INTEREST_LIST ||
             type == TcrMessage::REGISTER_INTEREST ||
             type == TcrMessage::QUERY ||
             type == TcrMessage::QUERY_WITH_PARAMETERS ||
             type == TcrMessage::GET_ALL_70 ||
             type == TcrMessage::GET_ALL_WITH_CALLBACK ||
             type == TcrMessage::PUTALL ||
             type == TcrMessage::PUT_ALL_WITH_CALLBACK ||
             type == TcrMessage::REMOVE_ALL ||
             ((type == TcrMessage::EXECUTE_FUNCTION ||
               type == TcrMessage::EXECUTE_REGION_FUNCTION) &&
              (request.hasResult() & 2)) ||
             type ==
                 TcrMessage::EXECUTE_REGION_FUNCTION_SINGLE_HOP ||  // This is
                                                                    // kept
                                                                    // aside as
                                                                    // server
                                                                    // always
                                                                    // sends
                                                                    // chunked
                                                                    // response.
             type == TcrMessage::EXECUTECQ_MSG_TYPE ||
             type == TcrMessage::STOPCQ_MSG_TYPE ||
             type == TcrMessage::CLOSECQ_MSG_TYPE ||
             type == TcrMessage::KEY_SET ||
             type == TcrMessage::CLOSECLIENTCQS_MSG_TYPE ||
             type == TcrMessage::GETCQSTATS_MSG_TYPE ||
             type == TcrMessage::MONITORCQ_MSG_TYPE ||
             type == TcrMessage::EXECUTECQ_WITH_IR_MSG_TYPE ||
             type == TcrMessage::GETDURABLECQS_MSG_TYPE) {
    sendRequestForChunkedResponse(request, reply, conn);
    LOGDEBUG("sendRequestConn: calling sendRequestForChunkedResponse DONE");
  } else {
    // Chk request type to request if so request.getCallBackArg flag & setCall
    // back arg flag to true, and in response chk for this flag.
    if (request.getMessageType() == TcrMessage::REQUEST) {
      if (request.isCallBackArguement()) {
        reply.setCallBackArguement(true);
      }
    }
    size_t dataLen;
    LOGDEBUG("sendRequestConn: calling sendRequest");
    auto data = conn->sendRequest(request.getMsgData(), request.getMsgLength(),
                                  &dataLen, request.getTimeout(),
                                  reply.getTimeout(), request.getMessageType());
    reply.setMessageTypeRequest(type);
    reply.setData(
        data, static_cast<int32_t>(dataLen), this->getDistributedMemberID(),
        *(m_cacheImpl->getSerializationRegistry()),
        *(m_cacheImpl
              ->getMemberListForVersionStamp()));  // memory is released by
                                                   // TcrMessage setData().
  }

  // reset idle timeout of the connection for pool connection manager
  if (type != TcrMessage::PING) {
    conn->touch();
  }

  if (reply.getMessageType() == TcrMessage::INVALID) {
    if (type == TcrMessage::EXECUTE_FUNCTION ||
        type == TcrMessage::EXECUTE_REGION_FUNCTION ||
        type == TcrMessage::EXECUTE_REGION_FUNCTION_SINGLE_HOP) {
      ChunkedFunctionExecutionResponse* resultCollector =
          dynamic_cast<ChunkedFunctionExecutionResponse*>(
              reply.getChunkedResultHandler());
      if (resultCollector->getResult() == false) {
        LOGDEBUG("TcrEndpoint::send: function execution, no response desired");
        //            m_opConnections.put( conn, false );
        //  return GF_NOERR;
        error = GF_NOERR;
      }
    } else {
      // Treat INVALID messages like IO exceptions
      error = GF_IOERR;
    }
  }
  // do we need to consider case where compareTransactionIds return true?
  // I think we will not have issue here
  else if (!compareTransactionIds(request.getTransId(), reply.getTransId(),
                                  failReason, conn)) {
    error = GF_NOTCON;
  }
  if (error == GF_NOERR) {
    if (m_baseDM != nullptr)
      m_baseDM->afterSendingRequest(request, reply, conn);
  }

  return error;
}

bool TcrEndpoint::isMultiUserMode() {
  LOGDEBUG("TcrEndpoint::isMultiUserMode %d", m_isMultiUserMode);
  return m_isMultiUserMode;
}

GfErrType TcrEndpoint::sendRequestWithRetry(
    const TcrMessage& request, TcrMessageReply& reply, TcrConnection*& conn,
    bool& epFailure, std::string& failReason, int maxSendRetries,
    bool useEPPool, std::chrono::microseconds requestedTimeout,
    bool isBgThread) {
  GfErrType error = GF_NOTCON;
  bool createNewConn = false;
  // int32_t type = request.getMessageType();
  int sendRetryCount = 0;

  //  Retry on the following send errors:
  // Timeout: 1 retry
  // EAGAIN, ECONNRESET, EWOULDBLOCK: 1 retry
  // Connection pool is empty (too many threads or no connections available): 1
  // retry

  do {
    if (sendRetryCount > 0) {
      // this is a retry. set the retry bit in the early Ack
      (const_cast<TcrMessage&>(request)).updateHeaderForRetry();
    }

    auto timeout = requestedTimeout;
    epFailure = false;
    if (useEPPool) {
      if (m_maxConnections == 0) {
        ACE_Guard<ACE_Recursive_Thread_Mutex> guard(m_connectionLock);
        if (m_maxConnections == 0) {
          LOGFINE(
              "Creating a new connection when connection-pool-size system "
              "property set to 0");
          if ((error = createNewConnection(conn, false, false,
                                           m_cacheImpl->getDistributedSystem()
                                               .getSystemProperties()
                                               .connectTimeout())) !=
              GF_NOERR) {
            epFailure = true;
            continue;
          }
          m_maxConnections = 1;
        }
      }
    }
    LOGDEBUG("TcrEndpoint::send() getting a connection for endpoint %s",
             m_name.c_str());
    if (createNewConn) {
      createNewConn = false;
      if (!m_connected) {
        return GF_NOTCON;
      } else if ((error =
                      createNewConnection(conn, false, false,
                                          m_cacheImpl->getDistributedSystem()
                                              .getSystemProperties()
                                              .connectTimeout(),
                                          0, true)) != GF_NOERR) {
        epFailure = true;
        continue;
      }
    } else if (conn == nullptr && useEPPool) {
      LOGFINER(
          "sendRequestWithRetry:: looking for connection in queue timeout = "
          "%d ",
          timeout.count());
      // max wait time to get a connection
      conn = m_opConnections.getUntil(timeout);
    }
    if (!m_connected) {
      return GF_NOTCON;
    }
    if (conn != nullptr) {
      LOGDEBUG("TcrEndpoint::send() obtained a connection for endpoint %s",
               m_name.c_str());
      int reqTransId = request.getTransId();

      try {
        LOGDEBUG("Calling sendRequestConn");
        error = sendRequestConn(request, reply, conn, failReason);
        if (error == GF_IOERR) {
          epFailure = true;
          failReason = "received INVALID reply from server";
          if (!handleIOException(failReason, conn, isBgThread)) {
            break;
          }
          createNewConn = true;
        } else if (error == GF_NOTCON) {
          epFailure = true;
          createNewConn = true;
        } else {
          if (useEPPool) {
            m_opConnections.put(conn, false);
          }
          return GF_NOERR;
        }
      } catch (const TimeoutException&) {
        error = GF_TIMOUT;
        LOGFINE(
            "Send timed out for endpoint %s. "
            "Message txid = %d",
            m_name.c_str(), reqTransId);
        closeFailedConnection(conn);
        /*
        if ( !(m_poolHADM && m_poolHADM->getThreadLocalConnections()) ){ //close
        connection only when not a sticky connection.
          closeConnection( conn );
        }*/
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
        int32_t type = request.getMessageType();
        epFailure = (type != TcrMessage::QUERY && type != TcrMessage::PUTALL &&
                     type != TcrMessage::PUT_ALL_WITH_CALLBACK &&
                     type != TcrMessage::EXECUTE_FUNCTION &&
                     type != TcrMessage::EXECUTE_REGION_FUNCTION &&
                     type != TcrMessage::EXECUTE_REGION_FUNCTION_SINGLE_HOP &&
                     type != TcrMessage::EXECUTECQ_WITH_IR_MSG_TYPE);

        // epFailure = true;
        failReason = "timed out waiting for endpoint";
        createNewConn = true;
      } catch (const GeodeIOException& ex) {
        error = GF_IOERR;
        epFailure = true;
        failReason = "IO error for endpoint";
        if (!handleIOException(ex.what(), conn,
                               isBgThread)) {  // change here
          break;
        }
        createNewConn = true;
      } catch (const Exception& ex) {
        failReason = ex.getName();
        failReason.append(": ");
        failReason.append(ex.what());
        LOGWARN("Error during send for endpoint %s due to %s", m_name.c_str(),
                failReason.c_str());
        if (compareTransactionIds(reqTransId, reply.getTransId(), failReason,
                                  conn)) {
          if (Log::warningEnabled()) {
            LOGWARN("Stack trace: %s", ex.getStackTrace().c_str());
          }
          error = GF_MSG;
          if (useEPPool) {
            m_opConnections.put(conn, false);
          } else {
            // we are here its better to close the connection as
            // "compareTransactionIds"
            // will not close the connection
            closeConnection(conn);
          }
          break;
        } else {
          error = GF_NOTCON;
          epFailure = true;
          createNewConn = true;
        }
      } catch (...) {
        failReason = "unexpected exception";
        LOGERROR(
            "Unexpected exception while sending request to "
            "endpoint %s",
            m_name.c_str());
        if (compareTransactionIds(reqTransId, reply.getTransId(), failReason,
                                  conn)) {
          error = GF_MSG;
          if (useEPPool) {
            m_opConnections.put(conn, false);
          } else {
            // we are here its better to close the connection as
            // "compareTransactionIds"
            // will not close the connection
            closeConnection(conn);
          }
          break;
        } else {
          error = GF_NOTCON;
          epFailure = true;
          createNewConn = true;
        }
      }
    } else {
      if (useEPPool) {
        epFailure = true;
        failReason = "server connection could not be obtained";
        if (timeout <= std::chrono::microseconds::zero()) {
          error = GF_TIMOUT;
          LOGWARN(
              "No connection available for %ld seconds "
              "for endpoint %s.",
              requestedTimeout.count(), m_name.c_str());
        } else {
          error = GF_NOTCON;
          LOGFINE(
              "Returning without connection with %d seconds remaining "
              "for endpoint %s.",
              timeout.count(), m_name.c_str());
        }
      } else {
        LOGERROR("Unexpected failure while sending request to server.");
        GF_DEV_ASSERT("Bug in TcrEndpoint::sendRequestWithRetry()?" ? false
                                                                    : true);
      }
    }
  } while (++sendRetryCount <= maxSendRetries);
  return error;
}

void TcrEndpoint::setRetry(const TcrMessage& request, int& maxSendRetries) {
  int32_t type = request.getMessageType();
  if (type == TcrMessage::QUERY || type == TcrMessage::QUERY_WITH_PARAMETERS ||
      type == TcrMessage::PUTALL || type == TcrMessage::PUT_ALL_WITH_CALLBACK ||
      type == TcrMessage::EXECUTE_FUNCTION ||
      type == TcrMessage::EXECUTE_REGION_FUNCTION ||
      type == TcrMessage::EXECUTE_REGION_FUNCTION_SINGLE_HOP ||
      type == TcrMessage::EXECUTECQ_WITH_IR_MSG_TYPE) {
    maxSendRetries = 0;
  }
}

GfErrType TcrEndpoint::send(const TcrMessage& request, TcrMessageReply& reply) {
  GfErrType error = GF_NOTCON;

  int maxSendRetries = 1;
  setRetry(request, maxSendRetries);

  TcrConnection* conn = nullptr;
  bool epFailure;
  std::string failReason;
  //  TODO: remove sendRetryCount as parameter.
  error = sendRequestWithRetry(request, reply, conn, epFailure, failReason,
                               maxSendRetries, true, reply.getTimeout());

  if (error == GF_NOERR) {
    m_msgSent = true;
  }

  if (error != GF_NOERR && epFailure) {
    LOGFINE("Send Giving up for endpoint %s; reason: %s.", m_name.c_str(),
            failReason.c_str());
    setConnectionStatus(false);
  }

// Postconditions:
#if GF_DEVEL_ASSERTS == 1
  int opConnectionsSize = m_opConnections.size();
  if (!m_isActiveEndpoint && (opConnectionsSize > 1)) {
    LOGWARN("Connections size = %d, expected maximum %d", opConnectionsSize, 1);
  } else if (opConnectionsSize > m_maxConnections) {
    LOGWARN("Connections size = %d, expected maximum %d", opConnectionsSize,
            m_maxConnections);
  }
#endif

  return error;
}

GfErrType TcrEndpoint::sendRequestConnWithRetry(const TcrMessage& request,
                                                TcrMessageReply& reply,
                                                TcrConnection*& conn,
                                                bool isBgThread) {
  GfErrType error = GF_NOTCON;

  int maxSendRetries = 1;
  setRetry(request, maxSendRetries);

  //  Retry on the following send errors:
  // Timeout: 1 retry
  // EAGAIN, ECONNRESET, EWOULDBLOCK: 1 retry
  // Connection pool is empty (too many threads or no connections available): 1
  // retry
  bool epFailure;
  std::string failReason;
  LOGFINE("sendRequestConnWithRetry:: maxSendRetries = %d ", maxSendRetries);
  error = sendRequestWithRetry(request, reply, conn, epFailure, failReason,
                               maxSendRetries, false, reply.getTimeout(),
                               isBgThread);
  if (error == GF_NOERR) {
    m_msgSent = true;
  }

  if (error != GF_NOERR && epFailure) {
    LOGFINE("sendRequestConnWithRetry: Giving up for endpoint %s; reason: %s.",
            m_name.c_str(), failReason.c_str());
    setConnectionStatus(false);
  }

  return error;
}

void TcrEndpoint::setConnectionStatus(bool status) {
  // : Store the original value of m_isActiveEndpoint.
  // This is to try make failover more resilient for the case when
  // a foreground operation thread is connecting to an endpoint while
  // the notification thread is disconnecting from the same, or vice versa.
  // By comparing the original value with the new value we know if
  // someone else has changed the status in that duration, and skip
  // the change if that is the case.
  // Same logic applies for the ping thread.
  // Try something like (after the 2.5 patch release):
  // bool wasActive = m_isActiveEndpoint;
  // Then after taking the lock:
  // If ( !wasActive && isActiveEndpoint ) { return; }
  ACE_Guard<ACE_Recursive_Thread_Mutex> guard(m_connectionLock);
  if (m_connected != status) {
    bool connected = m_connected;
    m_connected = status;
    if (connected) {
      m_numberOfTimesFailed += 1;
      m_isAuthenticated = false;
      // disconnected
      LOGFINE("Disconnecting from endpoint %s", m_name.c_str());
      closeConnections();
      m_isActiveEndpoint = false;
      LOGFINE("Disconnected from endpoint %s", m_name.c_str());
      triggerRedundancyThread();
    }
  }
}

void TcrEndpoint::triggerRedundancyThread() {
  m_failoverSema.release();
  m_redundancySema.release();
}

void TcrEndpoint::closeConnection(TcrConnection*& conn) {
  conn->close();
  m_ports.erase(conn->getPort());
  _GEODE_SAFE_DELETE(conn);
}

void TcrEndpoint::closeConnections() {
  m_opConnections.close();
  m_ports.clear();
  m_maxConnections = m_cacheImpl->getDistributedSystem()
                         .getSystemProperties()
                         .connectionPoolSize();
}

/*
void TcrEndpoint::sendNotificationCloseMsg()
{
  if (m_notifyConnection != nullptr) {
    m_notifyReceiver->stop();
    m_notifyConnection->close();
  }
}
*/

void TcrEndpoint::closeNotification() {
  LOGFINEST("Closing subscription channel for endpoint %s", m_name.c_str());
  m_notifyConnection->close();
  m_notifyReceiver->stopNoblock();
  TcrConnectionManager& tccm = m_cacheImpl->tcrConnectionManager();
  tccm.addNotificationForDeletion(m_notifyReceiver, m_notifyConnection,
                                  m_notificationCleanupSema);
  m_notifyCount++;
  m_cleanupSema.release();
  m_isQueueHosted = false;
  LOGFINEST(
      "Added susbcription channel for deletion and "
      "released cleanup semaphore for endpoint %s",
      m_name.c_str());
}

void TcrEndpoint::stopNoBlock() {
  if (m_notifyReceiver != nullptr) {
    m_notifyConnection->close();
    m_notifyReceiver->stopNoblock();
  }
}

void TcrEndpoint::stopNotifyReceiverAndCleanup() {
  LOGFINER("Stopping subscription receiver and cleaning up");
  ACE_Guard<ACE_Recursive_Thread_Mutex> guard(m_notifyReceiverLock);

  if (m_notifyReceiver != nullptr) {
    LOGFINER("Waiting for notification thread...");
    // m_notifyReceiver->stopNoblock();
    m_notifyReceiver->wait();
    bool found = false;
    for (std::list<Task<TcrEndpoint>*>::iterator it =
             m_notifyReceiverList.begin();
         it != m_notifyReceiverList.end(); it++) {
      if (*it == m_notifyReceiver) {
        found = true;
        break;
      }
    }

    if (!found) {
      _GEODE_SAFE_DELETE(m_notifyReceiver);
      _GEODE_SAFE_DELETE(m_notifyConnection);
    }
  }

  m_numRegionListener = 0;

  if (m_notifyReceiverList.size() > 0) {
    LOGFINER("TcrEndpoint::stopNotifyReceiverAndCleanup: notifylist size = %d",
             m_notifyReceiverList.size());
    for (std::list<Task<TcrEndpoint>*>::iterator it =
             m_notifyReceiverList.begin();
         it != m_notifyReceiverList.end(); it++) {
      LOGFINER(
          "TcrEndpoint::stopNotifyReceiverAndCleanup: deleting old notify "
          "recievers.");
      _GEODE_SAFE_DELETE(*it);
    }
  }

  if (m_notifyConnectionList.size() > 0) {
    LOGFINER("TcrEndpoint::stopNotifyReceiverAndCleanup: notifylist size = %d",
             m_notifyConnectionList.size());
    for (std::list<TcrConnection*>::iterator it =
             m_notifyConnectionList.begin();
         it != m_notifyConnectionList.end(); it++) {
      LOGFINER(
          "TcrEndpoint::stopNotifyReceiverAndCleanup: deleting old notify "
          "connections.");
      _GEODE_SAFE_DELETE(*it);
    }
  }
}

void TcrEndpoint::setServerQueueStatus(ServerQueueStatus queueStatus,
                                       int32_t queueSize) {
  if (!m_isServerQueueStatusSet) {
    m_isServerQueueStatusSet = true;
    m_serverQueueStatus = queueStatus;
    m_queueSize = queueSize;
  }
}

bool TcrEndpoint::isQueueHosted() { return m_isQueueHosted; }

void TcrEndpoint::processMarker() {
  m_cacheImpl->tcrConnectionManager().processMarker();
}

std::shared_ptr<QueryService> TcrEndpoint::getQueryService() {
  return m_cacheImpl->getQueryService(true);
}

void TcrEndpoint::sendRequestForChunkedResponse(const TcrMessage& request,
                                                TcrMessageReply& reply,
                                                TcrConnection* conn) {
  conn->sendRequestForChunkedResponse(request, request.getMsgLength(), reply);
}

void TcrEndpoint::closeFailedConnection(TcrConnection*& conn) {
  closeConnection(conn);
}

void TcrEndpoint::handleNotificationStats(int64_t) {}

}  // namespace client
}  // namespace geode
}  // namespace apache

redpajama_set_name: RedPajamaGithub


The item Digimon digital monsters, The official 2nd season, Vol. 6, Toei Animation presents ; original author, Akiyoshi Hongo ; producers, Daisuke Kawakami, Kyotaro Kimura, Hiromi Seki ; series director, Hiroyuki Kakudou represents a specific, individual, material embodiment of a distinct intellectual or artistic creation found in Salt Lake County Library Services.
Digifans, catch the final installment of Digimon Adventures in this second season. With some stellar teamwork by Davis, Yolei, Cody, T.K and Kari, the Digital World once again settles into a state of peace. The new Digidestined crew welcomes Ken back from his dark days as an evil Emperor and together they commit to protecting the Digital World and all its Digimon. However, the calm does not last long. Soon dark energy runs out of control and threatens to explode!


text:
DINO NO-NO
T-Rex Wreck: Mongolian Representative Disrupts Skeleton Auction
By Laura L. Griffin • 05/21/12 8:30am
Everyone expected the auction of the T-Rex skeleton to go peacefully. But they were wrong.
On the fourth floor of an airy gallery building in Chelsea on Sunday afternoon, a small group of private collectors and natural history enthusiasts gathered to witness the historic auction of a 24-foot-long, museum-quality Tyrannosaurus bataar skeleton, which was expected to bring up to $1.5 M. After a couple hours of uneventful sales of things like fine gemstones and amber-trapped insects, the marquee object came up.
The centerpiece auction was abruptly interrupted.
Before bidding on the skeleton began, the auctioneer announced that the sale would be "contingent upon a court proceeding dealing with this matter." Almost immediately, Robert Painter, a lawyer representing Elbegdorj Tsakhia, the president of Mongolia, stood up with a cell phone held to his ear and yelled, "I'm sorry, I need to interrupt this auction. I have a judge on the phone."
A loud whistle rang out, and Heritage Auctions president Greg Rohan, along with a group of security guards, gathered around Mr. Painter. They urged him to a corner of the room, past a massive cast of a sabre-tooth tiger and a glass case full of meteorite chunks.
In the past week, many prominent paleontologists have taken issue with the provenance of the skeleton and the legality with which it ended up at the Heritage auction. News of the sale eventually reached the Mongolian head-of-state, and Mr. Painter says he was contacted by President Elbegdorj's office "Friday after 5 PM" to see what he could do to postpone the auction.
The next day, Mr. Painter told The Observer, he filed a lawsuit against Heritage Auctions in Dallas, Texas (the headquarters of the auction house), alleging the skeleton was taken from Mongolia, and that, under Mongolian criminal law, "the export of dinosaur bones and fossils is a criminal offense."
Mr. Painter also got a temporary restraining order signed by district court judge Carlos Cortez, who he had on the phone during the auction, which was intended to prevent the sale and transfer of possession of the skeleton. (Mr. Painter provided us copies of both.)
"I was quite surprised and disappointed that Heritage Auctions chose knowingly to ignore a court order that the auction could not proceed," Mr. Painter said.
Despite the ruckus, the auction continued, and the considerable artifact sold for $1,052,500, to an unidentified phone bidder. The small audience, slightly confused, applauded.
After being escorted from the building, Mr. Painter stood with a small gathering of protestors—one of them wearing a traditional Mongolian hat and jacket—who were distributing fliers on the sidewalk. They held a sign that read, rather dispassionately, "It's a national treasure of Mongolia. Return our stolen treasure." Also among them was prominent Mongolian paleontologist Bolortsetseg Minjin, a representative of the Mongolian Academy of Sciences and the director of the Institute for the Study of Mongolian Dinosaurs, and Ann Altman, of Connecticut, who represented of one of the Mongolian president's advisors.
Ms. Altman explained, "We wanted postponement and clarification [of the provenance of the skeleton]. The president has made this his personal mission. He personally has stepped in to protect the patrimony of the country."
The protestors had learned of the sale only days prior, in an article in the Daily Mail, which states the origin of the skeleton was Mongolia.
Mark Norell, the chairman and curator of the division of paleontology at the American Museum of Natural History wrote a letter in support of the Mongolians, included in Mr. Painter's affidavit, which says the skeleton "clearly [was] excavated in Mongolia as this is the only locality in the world where these dinosaurs are known," and that the specimen was "undoubtedly looted from Mongolia."
But the auction catalog provided by Heritage Auctions does not so clearly state the origin of the skeleton. It notes that the skeleton is from "Central Asia" and that the Tyrannosaurus bataar lived in what is now the Gobi Desert (half of which lies in Mongolia), but does not provide further information.
In a brief interview, Mr. Rohan, the president of Heritage Auctions, told The Observer he was confident that the transport of the skeleton to the U.S. was legal. "We and our counsel are not aware of any treaties under which the sale would be illegal," Mr. Rohan said.
An updated press release on the Heritage Auction website noted that Heritage Auctions
fully appreciate[s] the widespread concerns relating to the Tyrannosaurus, but our conclusion is that no impropriety exists to prevent its sale at auction. Our consignor is an individual with a good reputation who has warrantied in writing to us that he holds clear title to the specimen. We've seen no evidence even suggesting that the fossils were collected illegally.
Mr. Rohan also discussed the complex patrimony issues that arise when dealing artifacts. "In a perfect world, the scientific community would like the sale of fossils to be illegal, and they would all be in museums. But the fact is it is legal to sell fossils. We as the auction house remain neutral [on this issue]. We represent the consignor, who in this case spent a year of his life repairing the skeleton," Mr. Rohan said.
He also said he felt that the negative attention in the press had "kept bidders away."
In the hours after the auction, Heritage posted another press release to its website, maintaining that the skeleton was acquired legally and denying Mongolia's claims to the artifact:
As far as we know, the Mongolian government has not produced any evidence that the piece originated in its territory, but the final determination will be up to the American legal system.
Mr. Painter plans to ask the judge for a shortened discovery period, with the hope that it will take a week rather than the typical month to get this ironed out.
"I feel comfortable that the Tyrannosaurus skeleton is safe. Once we get the ownership identified, if we're correct and this is the property of the Mongolian nation, this will not go back to the consignor—it will be returned to the Mongolian people. That's our goal," he said.
The buyer of the bones bid by phone, and his or her name was not disclosed by the auction house.
Filed Under: Politics, News & Politics, Auctions, Heritage Auctions, Dinosaurs, Mongolia, Patrimony, DINO NO-NO
SEE ALSO: Juul Labs Brings on Top Trumpworld Talent as Federal Investigators Circle

redpajama_set_name: RedPajamaCommonCrawl


Joshua Benton is director of the Nieman Journalism Lab at Harvard University. The Lab is an attempt to figure out a productive future for quality journalism online. While an IRP fellow, he worked as a staff writer and columnist for The Dallas Morning News, where he spent seven years. He has reported from ten foreign countries and three times been a finalist for the Livingston Award for International Reporting. He previously worked at The Blade in Toledo. A native of small-town south Louisiana, he received a Bachelor of Arts degree in history from Yale University, where he served as editor-in-chief of the Yale Herald.


text:
David Judge
David is a playwright, performance poet and actor.
His most recent play SPARKPLUG, which David wrote and performed, toured the UK in Spring 2019 and was shown on Sky Arts in May 2021.
David has written a new play, CHATBACK, for the National Theatre Connections programme, and he is working with Sky Studios on an in-house tv drama series developed by and for new writers with producer Vicky Wharton. David has written his first episode of CORONATION STREET which was broadcast in June 2021 and his second episode airs in February 2023.
His first play SKIPPING ROPE was shortlisted for the Alfred Fagan Award and PANLID was commissioned by Talawa and the Royal Exchange.
As an actor David played Tybalt in the National Theatre/Sky Arts feature film of ROMEO AND JULIET and most recently played Lysander in Manchester's Homeground's theatre production of MIDSUMMER NIGHT'S DREAM directed by Sean Holmes.
Download David's CV
Visit David's IMDb


I know a lot of my readers are dealing with compulsive overeating/binge eating, and also depression. From time to time, I like to share posts that offer support and advice for recovery. At the very least, please know that you are never alone.
A nothing day is a day where you accomplish nothing positive (and no, watching 8 episodes of Orange is the New Black doesn't count.) There have been plenty of days where I binged and literally did nothing but eat. The feeling I had at the end of these days was one of complete and utter failure. I went to bed those nights feeling ashamed, pathetic, and weak. Guess what those feelings did? Contributed to my depression.
Even when you're depressed and feeling weak, you can usually muster the motivation or courage to do something positive. It can be something as simple as taking a shower (hey, sometimes that's all we've got in us) or going for a walk. Return a phone call or an email that you've meaning meaning to write. Just one thing so that at the end of the day you can know you did at least something to help your situation. Any step toward the light is better than standing still in the dark.
Once you start stringing enough something days together, you'll start to see and feel some progress. Eventually, this should help create a new feedback loop—one of motivation and progress. And even if you continue to struggle, at least you've accomplished something.
Like this post? Take a look at my other weight loss posts.
Thank you for posting this. Perfect timing, really, as I've been in a bit of a slump lately. "No more nothing day"… definitely adding that one to my little toolbox.
So true — I just learned recently that lazy Netflix watching Sundays at home are really not good for me (as much as I love them) … I need to be productive or I get the urge to eat junk, and then feel bad about it… no more laying on the couch all day for me!
Thank you so much for this post. After struggling with depression for the last eight months, I would wholeheartedly agree that doing just one thing a day is the most important thing to do. It is so so so hard in the moment even to take a ten minute walk or write an email. And sometimes the pain of doing it lasts the entire time you're doing that act. For me, even a simple task meant I would have to take a nap after. But for anyone going through it, just remember that it will slowly and imperceptibly get better every single day, especially if you take the advice in this post. So keep going! You are not alone, and you will get there.
Thank you, Andi, for all of your beautifully written posts, and your inspiring and touching memoir.
Oh Andie…When you write you speak to all of us, you bare it all and we see ourselves. I love this blog, I loved your book ( its right next to my chair right now while I type this, it STAYS there! ) and you inspire me. Keep on talking, I am LISTENING and FOLLOWING!!!
As so many others have stated, this post touched my heart. I have had those days. Thank you for saying that somedays the only positive thing that can be done is a shower. Your blog is beautiful & the food you prepare always looks delicious. I can't wait to read your book.
I needed this today. Spent all day yesterday on the couch eating handful after handful of trailmix and not having the energy to do anything on my to do list. Considering it a hiccup. Back to reality now and being mindful. Man!!! Being a responsible, healthy human being is hard! Thank you for what you do and for sharing your life, hiccups and all!
Yeah, I have the opposite problem, when I get depressed and up set (like now) I can't eat…very frustrating.
Thanks for your insight Andie!
I love how you reach so many people!!! You are changing the world my friend!!
It's so obvious that you moonlight as a mind-reader ;) Thank you for this post…I tend to have do nothing days on Sundays and I ALWAYS regret them. I find that forcing myself to get up out of bed in the morning (as opposed to telling myself I can lay in bed for the remainder of the morning) to take a walk helps me out of the fog. Thank you for your book, your blog, and for being you!
Thank you for this – great timing, especially today (I read this on a Monday). Love it! Thank you.
You don't have to get it all done, just get something done…it all counts.
Wonderful advice! And well needed today. thank you!
so incredibly true… I completely agree that on those nothing days (of which I've had more of recently than in a long time) I feel so much worse the next day and it becomes that much harder to come back to myself…positive acts. that is the difference.
How did I miss this post? This is the best advice I got in a long time. Currently I'm having a bad phase and this really seems like something I could do to get back to my happier self. Thank you so much Andie!
Andie is that your quote? " Any step toward the light is better than standing still in the dark." I must share it but I want to give credit where credit is due. Please advise. Brilliant!
Ha yes it is :) Thank you!
I came across this today, and it really articulated what I tend to do to myself. A good reminder to take small steps back in the right direction. Thanks!
Such good advice Andie. Things often do snowball and it would be good if they snowballed the other direction once in awhile..
I needed this today for sure. Thanks Andie.
Andie – Thank you! Your post this morning helped me remember that I am the only one that can pull myself out of this… depression is definitely a downward spiral and it's easy to deny. Today I'm going to keep looking for that one thing I can do to keep myself either level or moving up, and stop that downward spiral. I love your authenticity, thank you!
Thank you. Would love to read more of your thoughts on dealing with depression.


Complex solutions with use of rail, road and water transport. Complete organization of logistical chain, in country and abroad.
We specialize in the implementation of the most complicated orders, which makes a really big impression! However, in addition to our services, we also want to share other events from the world of DAHER. The most interesting presented below.
The construction of new power units 5 and 6 at the Opole Power Plant is one of the largest infrastructure projects in Europe.
Recent projects of DAHER PROJECTS took place not only in Poland or Europe but also in the USA and Canada.
We invite all interested in our offer. DAHER PROJECTS SP. z o. o., are not only the highest quality transport services of loads of large dimensions and weights, but also professional service backed by 150 years of experience. We can realize any, even the most difficult project. Our engineers are always at your disposal and will be happy to provide assistance and professional advice.
Professional execution of orders from "A" to Z "
DAHER PROJECTS Sp. o.o. is a comprehensive logistic partner in cargo transport of large dimensions and masses, specializing in road and rail transport. We have a own vehicle park and also mediate in arranging the necessary permits and insurances. Many years of experience and deep expertise guarantees the professional execution of individual orders from ""A" to "Z" to". Really, nothing is impossible in the transport of extreme loads, provided that it performs DAHER PROJECTS SP. z o. o. . . .


With so much being written about content marketing it is easy to understand how the amount of misinformation out there almost outweighs the helpful stuff. To help you distinguish the good from the bad, here are five myths that may be hurting your bottom line.
The truth is that everyone should be involved in the content but only a professional writer should craft the piece. Sharing important knowledge across the organization includes bringing in perspectives on audience definition, product offerings and sales and marketing. It is up to the writer to do the research required to craft an engaging and informative piece that provides value and speaks to the intended audience.
If this were true, would Cisco and GE have launched some seriously engaging story campaigns? Probably not. Who would have thought that Cisco could make a video that was so entertaining or that GE would have more than 25k Pinterest followers?
This has everything to do with story. At the end of the day, it is people that make the decision to buy, not brands or businesses; and people love stories.
Ever hear of a little company called Hubspot? Measuring the impact of your content marketing efforts requires setting quantifiable goals and then tracing back your content efforts to understand how they impact your success. It can get a little complicated which why marketing automation software, like Hubspot, is essential.
It actually is easy to find cheap writers. Just add a quick job post on any freelancing site and you will have hundreds of applicants willing to produce content for as little as $2 a post; 300 to 600 words for less than a cup of coffee. You should check it out… I dare you. But remember, you get what you pay for.
"Hey, wait until you read this great post all about Brand X's features and benefits…and next week they are going to post on some internal staffing changes!" Have you ever received an email like that? Me neither. Your customer and potential customers do not want to read about your product on your blog; that is why you have product pages. Nor do they want to hear about company events, save those for the employee newsletter. Your blog is a place for you to let your audience know how you feel about things other than your product. To show them that there is an entire set of values and interests that round-out the company and make it a really cool partner. For a great example of social outreach on steroids, check out American Express' OPEN forum; this 160 year old company has nailed the whole social media/content marketing conundrum in one intelligently designed, expertly crafted site.
With all of the information on content marketing that pervades the internet it can be extremely difficult to separate myth from reality. Avoid these pitfalls and only post content that is truly valuable to your audience.
Is Your Content Turning a Profit?


From the start, we've known that JJ Abrams' Super 8 was heavily inspired by Steven Spielberg's Amblin-era films. I didn't want to try to compile a listing of the many possible homages, as some of them could be very subjective and not intentional by the filmmaker. I thought it would be better to focus on the hidden cameos, inside jokes, hidden names and so on.
Downtown there is a store called "James Locke Jeweler", a reference to Terry O'Quinn's character John Locke on the JJ produced television series Lost.
Another possible Lost reference is the name of the main character's father: Jack Lamb is very simular to Lost's lead character Jack Shephard. Lamb, Shephard?
Greg Grunberg, best known for his role as Matt Parkman on Heroes, has been friends with Abrams since kindergarten. Grunberg starred in all of the young filmmaker's home movies growing up, and JJ has tried to include Grunberg in every one of his movies and tv shows as a "lucky charm". Grunberg co-starred in Felicity as Sean Blumberg, co-starred in Alias as Eric Weiss, played the plane's pilot, the smoke monster's first victim, in Lost, had a small role as "Kevin" in Abrams' feature debut Mission: Impossible III. Grunberg was too busy on the set of Heroes to appear in Star Trek, but he can be heard as the voice of James T Kirk's Stepdad on the cellphone call in the early car sequence.
Grunberg revealed to the Associated Press that people will have to "really, really search to find me" as he portrays a sitcom actor in the movie. You can hear Greg say something like "I always wear shirts with lipstick on the collar!", which comes from a television set above as Joe first enters the alien creatures den. You can see the tv in the background, but I'm not sure if you can see Greg.
Abrams' longtime composer Michael Giacchino also makes an appearance in the film. My friend Jordan Raup informs me that Giacchino appears as Deputy Crawford, the police officer that Chandler makes fun of for not knowing how the radio works.
We have so many easter egg photos that we couldn't fit them on just one page….


Push right function key once. It should say find channels.
Click on the scroll right arrow until it says your ID. Click enter. Type in your student ID number without the dash. Press Enter.
Your device should now be set up for use in class.
Your ID will now be registered with the clicker and will be registered, along with your responses, each time you use the clicker in class. You will be awarded clicker points toward your quiz score for each question as follows: 0% for no answer, 50% for incorrect answer, 100% for correct answer. I recommend that you check to be sure that you are receiving points and let me know ASAP if not the case.
Batteries: Each device has a battery-level indicator is illuminated each time the device is turned on. The batteries that came with your PRS (2-AAA) may last for the entire semester, but they could also fail unexpectedly. Turn your clicker off when it is not in use, especially if it is being bounced around in your backpack.
Lost/Forgot Clicker: In the case that your clicker is lost or stolen, students are responsible for providing a replacement before the next class. In this case, the instructor must be notified so the new unit can be logged in and there is no loss of information. There are no replacements available during class for forgotten, lost or stolen clickers.
If these or other problems arise, please let me know immediately following class.


text:
La Charte des devoirs professionnels des journalistes français a été initialement rédigée en , révisée en , puis en . À l'occasion de cette dernière révision, elle a été rebaptisée Charte d'éthique professionnelle des journalistes. Cette charte n'implique pas d'obligation légale.

Histoire
Le Syndicat national des journalistes est à l'origine des trois versions successives de ce texte, actualisées à chaque fois à l'issue du vote d'un congrès, en 1918, en 1938 et en 2011.

La version la plus récente, adoptée en 2011, reprend le principe du secret professionnel, déjà présent en 1918 et en 1938, en y ajoutant un devoir jugé essentiel, celui de protection des sources d'information des journalistes. Alors que les versions de 1918 et 1938 tenaient en une phrase de quatre mots (), la version de 2011 y ajoute sept mots pour écrire au total .

Une démarche identique avait déjà été adoptée en 1971 par la Charte de Munich. Son article 7 reprenait les quatre mots de la phrase de 1918 et 1938  mais y ajoutait .

La version 2011 de la charte du Syndicat national des journalistes est encore plus protectrice, puisqu'elle ne limite plus la protection des sources d'information à celles ayant fourni des informations « confidentiellement », mais à toutes les sources d'information, dans un souci de confiance.

La version de 1938 indique que le journaliste  Dans la version 2011, il n'est plus question que de proscrire . Les émissions télévisées comme Les Infiltrés sur France 2, ou un livre comme Tête de Turc de Günter Wallraff, se voient ainsi devenir conformes à la charte.

Notes et références

Voir aussi

Articles connexes
 Déontologie du journalisme

Liens externes
 Charte de 1918, remaniée en 1938, sur le site internet du Syndicat national des journalistes ;
 Charte de 2011, sur le site internet du Syndicat national des journalistes.

Journalisme en France
Code de conduite

redpajama_set_name: RedPajamaWikipedia


Kerala has been authorized as one of the ten heavens on Earth by National Geographic. The demography and the expanded social foundation of Kerala is a perfect mix for an occasion goal. Kerala packages typically include visit to Lush slope stations, Sandy shorelines, Religious, Heritage and Cultural sights, Ayurveda treatment. Kerala has been named reasonably as The Gods' Own Country & Venice of East.
The Venice of the East is filled with beautiful magnificence of nature & prosperous culture & custom. It is a position of deep sense of spirituality and grandeur. A vacationer can have the delight of visit through a scope of tour packages which offer various choices to investigate the prosperous custom & culture & beautiful characteristic excellence of Kerala. In Kerala a traveler can make visits to destinations, for example, Trivandrum, Munnar, Alappuzha, Thekkady, and Palakkad and so on. All these places in Kerala are notable for their rich charisma and culture and stuffed with bunches of awe-inspiring must see attractions.
When you pay visit to the peaceful backwaters amid your visits, the Venice of the East, the primary thing that gets your sight the Kerala houseboats. The houseboats of Kerala are entirely indistinguishable, the gondolas in the Venice. The greater part of the vacationers, particularly the wedding trip couples, love to remain overnight in an extravagance houseboat. It could be a truly phenomenal ordeal interestingly explorers in Kerala. The Kerala backwater tours are for the most part offered as Kottayam Backwater visit, Kozhikode Backwater visits, Alleppey Backwater packages; Kochi Backwater visits packages, Kumarakom Backwater visit packages.
The sandy sun-kissed delightful shorelines are additionally exceptionally celebrated among the traveler. The lovely coastline includes wealth of appealing and clean sandy shorelines, for example, Kappad, Kappil, Kovalam , Alappuzha , Cherai , Marari shoreline & Beypore Beach & so on. You can choose your decision of shoreline visit from the extensive variety of shoreline tours accessible in Kerala. Simply leave whatever is left of the world behind and appreciate without bounds & have a decent time on your vacation to the shorelines.


A workspace is a user-defined arrangement of panels. The user is able to open multiple panels and arrange them according to the user's preference. A workspace can be saved and opened so that the user does not have to keep re-arranging the panels every time. Each user usually tends to organize the views in a certain way to perform a specific task. By means of that feature, the application provides the facility to save those different configurations and recover them at will.


text:
HomeMoneyRare Kew Gardens 50p coin sells for £250 on eBay – how...
Rare Kew Gardens 50p coin sells for £250 on eBay – how to spot one and make a mint
A RARE 50p coin has sold on eBay 500 times higher than its face value, as one lucky bidder managed to claim it for their own with a £250 offer.
In any other circumstance a 50p is worth exactly that – 50p. But this coin features a unique twist that means it's more in demand than any other.
The rare coin has the famous Kew Gardens design on the reverseCredit: ebay
The eBay listing attracted the attention of 28 bidsCredit: ebay
That means collectors, who know what they're looking for, will offer high prices to get their owns hands on a copy.
It's because the coin is the rarest of them all -according to Change Checker's latest scarcity index anyway.
It's the rare 2009 Kew Gardens 50p, of which only 210,000 exist.
On the reverse side it features an image of the Chinese pagoda that stands at the heart of the London attraction the coin is named after.
We often see this particular style coin sell for hundreds of pounds on eBay so it's no strange feat, and this particular coin managed to catch the attention of 28 bids.
Those were placed within the short time frame of just one day, though bids kicked off at a whopping £90 to start with.
RARE COINS AND VALUABLE NOTES
We've seen copies of the coin reach £190 in the past, and £177 before too, so £250 is an impressive price to call the final sale on.
How to spot one and make a mint
It's always worthing having a rifle through your own change to spot any unique designs – if you find one you could make a lot of money from it on eBay.
The Kew Gardens 50p, like other commemorative style coins, will usually sell for hundreds of pounds, so you could definitely make a mint.
Even the odd minting error raises interest too.
They'll all usually be produced in low numbers so demand from collectors is high.
If you look at other listings on eBay you'll be able to determine how much your own change could go for.
But you should also always keep in mind that on eBay a buyer could pull out, which means the coin won't have sold for the price it says it has.
Be cautious of fakes too, as they'll often show up online.
You can check your change with experts like Coin Hunter or The Royal Mint if you want to know if your change is worth what others are saying.
Rare coins and valuable notes – is yours worth a mint?
Royal Mint reveals new Winnie the Pooh and friends 50p coin