sync

2025-09-28 19:05:55 +08:00 · 2018-10-02 15:39:13 +02:00 · 2018-10-02 15:39:13 +02:00 · 979e35049a
commit 979e35049a
parent e4cad79820
11 changed files with 90 additions and 20 deletions
--- a/include/ulib/net/client/redis.h
+++ b/include/ulib/net/client/redis.h
@ -59,15 +59,13 @@
 * @brief UREDISClient is a wrapper to REDIS API
 */

-typedef void (*vPFcs)(const UString&);
+typedef void (*vPFcs)  (const UString&);
+typedef void (*vPFcscs)(const UString&,const UString&);

 class U_EXPORT UREDISClient_Base : public UClient_Base, UEventFd {
 public:

-   ~UREDISClient_Base()
-      {
-      U_TRACE_DTOR(0, UREDISClient_Base)
-      }
+   ~UREDISClient_Base();

   // RESPONSE

@ -735,14 +733,12 @@ public:
      return processRequest(U_RC_MULTIBULK, U_CONSTANT_TO_PARAM("UNSUBSCRIBE"), param, len);
      }

+   void unsubscribe(const UString& channel);                   // unregister the callback for messages published to the given channels
+   void   subscribe(const UString& channel, vPFcscs callback); //   register the callback for messages published to the given channels
+
   // define method VIRTUAL of class UEventFd

-   virtual int handlerRead() U_DECL_FINAL
-      {
-      U_TRACE_NO_PARAM(0, "UREDISClient_Base::handlerRead()")
-
-      U_RETURN(U_NOTIFIER_OK);
-      }
+   virtual int handlerRead() U_DECL_FINAL;

   virtual void handlerDelete() U_DECL_FINAL
      {
@ -774,6 +770,7 @@ protected:
   static ptrdiff_t diff;
   static UVector<UString>* pvec;
   static UREDISClient_Base* pthis;
+   static UHashMap<void*>* pchannelCallbackMap;

   UREDISClient_Base() : UClient_Base(U_NULLPTR)
      {
--- a/src/ulib/net/client/redis.cpp
+++ b/src/ulib/net/client/redis.cpp
@ -16,9 +16,20 @@

 uint32_t           UREDISClient_Base::start;
 ptrdiff_t          UREDISClient_Base::diff;
+UHashMap<void*>*   UREDISClient_Base::pchannelCallbackMap;
 UVector<UString>*  UREDISClient_Base::pvec;
 UREDISClient_Base* UREDISClient_Base::pthis;

+UREDISClient_Base::~UREDISClient_Base()
+{
+   U_TRACE_DTOR(0, UREDISClient_Base)
+
+   if (pchannelCallbackMap)
+      {
+      U_DELETE(pchannelCallbackMap)
+      }
+}
+
 // Connect to REDIS server

 void UREDISClient_Base::init()
@ -530,6 +541,68 @@ bool UREDISClient_Base::deleteKeys(const char* pattern, uint32_t len) // Delete
   U_RETURN(true);
 }

+// PUB/SUB (@see http://redis.io/pubsub)
+
+void UREDISClient_Base::unsubscribe(const UString& channel) // unregister the callback for messages published to the given channels
+{
+   U_TRACE(0, "UREDISClient_Base::unsubscribe(%V)", channel.rep)
+
+   if (pchannelCallbackMap == U_NULLPTR)
+      {
+      U_NEW(UHashMap<void*>, pchannelCallbackMap, UHashMap<void*>);
+      }
+   else
+      {
+      (void) pchannelCallbackMap->erase(channel);
+      }
+}
+
+void UREDISClient_Base::subscribe(const UString& channel, vPFcscs callback) // register the callback for messages published to the given channels
+{
+   U_TRACE(0, "UREDISClient_Base::subscribe(%V,%p)", channel.rep, callback)
+
+   if (pchannelCallbackMap == U_NULLPTR)
+      {
+      U_NEW(UHashMap<void*>, pchannelCallbackMap, UHashMap<void*>);
+      }
+
+   pchannelCallbackMap->insert(channel, (const void*)callback);
+}
+
+// define method VIRTUAL of class UEventFd
+
+int UREDISClient_Base::handlerRead()
+{
+   U_TRACE_NO_PARAM(0, "UREDISClient_Base::handlerRead()")
+
+   U_INTERNAL_ASSERT_POINTER(pchannelCallbackMap)
+
+   if ((clear(), UClient_Base::response.setEmpty(), UClient_Base::readResponse(U_SINGLE_READ)))
+      {
+      char prefix = UClient_Base::response[0];
+
+      if (prefix != U_RC_MULTIBULK)
+         {
+         err = (prefix == U_RC_ERROR ? U_RC_ERROR
+                                     : U_RC_ERR_PROTOCOL);
+
+         U_RETURN(false);
+         }
+
+      err = U_RC_OK;
+
+      processResponse();
+
+      UString channel = vitem[1];
+
+      vPFcscs callback = (vPFcscs) pchannelCallbackMap->at(channel); 
+
+      if (callback) callback(channel, vitem[2]);
+      }
+
+   U_RETURN(U_NOTIFIER_OK);
+}
+
 // DEBUG

 #if defined(U_STDCPP_ENABLE) && defined(DEBUG)
--- a/tests/examples/TSA/tsaserial
+++ b/tests/examples/TSA/tsaserial
@ -1 +1 @@
-040E
+0415
--- a/tests/examples/benchmark/FrameworkBenchmarks/ULib/docker/ulib-json.dockerfile
+++ b/tests/examples/benchmark/FrameworkBenchmarks/ULib/docker/ulib-json.dockerfile
@ -45,7 +45,7 @@ RUN tar xf ULib-${ULIB_VERSION}.tar.gz
 WORKDIR $IROOT/ULib-$ULIB_VERSION

 # AVOID "configure: error: newly created file is older than distributed files! Check your system clock"
-#RUN cp /src/* src/ulib/net/server/plugin/usp 
+RUN cp /src/* src/ulib/net/server/plugin/usp 
 RUN find . -exec touch {} \;

 RUN echo "userver {" >> $ULIB_ROOT/benchmark.cfg
--- a/tests/examples/benchmark/FrameworkBenchmarks/ULib/docker/ulib-json_fit.dockerfile
+++ b/tests/examples/benchmark/FrameworkBenchmarks/ULib/docker/ulib-json_fit.dockerfile
@ -45,7 +45,7 @@ RUN tar xf ULib-${ULIB_VERSION}.tar.gz
 WORKDIR $IROOT/ULib-$ULIB_VERSION

 # AVOID "configure: error: newly created file is older than distributed files! Check your system clock"
-#RUN cp /src/* src/ulib/net/server/plugin/usp 
+RUN cp /src/* src/ulib/net/server/plugin/usp 
 RUN find . -exec touch {} \;

 RUN echo "userver {" >> $ULIB_ROOT/benchmark.cfg
--- a/tests/examples/benchmark/FrameworkBenchmarks/ULib/docker/ulib-mongodb.dockerfile
+++ b/tests/examples/benchmark/FrameworkBenchmarks/ULib/docker/ulib-mongodb.dockerfile
@ -51,7 +51,7 @@ RUN tar xf ULib-${ULIB_VERSION}.tar.gz
 WORKDIR $IROOT/ULib-$ULIB_VERSION

 # AVOID "configure: error: newly created file is older than distributed files! Check your system clock"
-#RUN cp /src/* src/ulib/net/server/plugin/usp
+RUN cp /src/* src/ulib/net/server/plugin/usp
 RUN find . -exec touch {} \;

 RUN echo "userver {" >> $ULIB_ROOT/benchmark.cfg
--- a/tests/examples/benchmark/FrameworkBenchmarks/ULib/docker/ulib-mysql.dockerfile
+++ b/tests/examples/benchmark/FrameworkBenchmarks/ULib/docker/ulib-mysql.dockerfile
@ -45,7 +45,7 @@ RUN tar xf ULib-${ULIB_VERSION}.tar.gz
 WORKDIR $IROOT/ULib-$ULIB_VERSION

 # AVOID "configure: error: newly created file is older than distributed files! Check your system clock"
-#RUN cp /src/* src/ulib/net/server/plugin/usp
+RUN cp /src/* src/ulib/net/server/plugin/usp
 RUN find . -exec touch {} \;

 RUN echo "userver {" >> $ULIB_ROOT/benchmark.cfg
--- a/tests/examples/benchmark/FrameworkBenchmarks/ULib/docker/ulib-plaintext_fit.dockerfile
+++ b/tests/examples/benchmark/FrameworkBenchmarks/ULib/docker/ulib-plaintext_fit.dockerfile
@ -45,7 +45,7 @@ RUN tar xf ULib-${ULIB_VERSION}.tar.gz
 WORKDIR $IROOT/ULib-$ULIB_VERSION

 # AVOID "configure: error: newly created file is older than distributed files! Check your system clock"
-#RUN cp /src/* src/ulib/net/server/plugin/usp 
+RUN cp /src/* src/ulib/net/server/plugin/usp 
 RUN find . -exec touch {} \;

 RUN echo "userver {" >> $ULIB_ROOT/benchmark.cfg
--- a/tests/examples/benchmark/FrameworkBenchmarks/ULib/docker/ulib-postgres.dockerfile
+++ b/tests/examples/benchmark/FrameworkBenchmarks/ULib/docker/ulib-postgres.dockerfile
@ -45,7 +45,7 @@ RUN tar xf ULib-${ULIB_VERSION}.tar.gz
 WORKDIR $IROOT/ULib-$ULIB_VERSION

 # AVOID "configure: error: newly created file is older than distributed files! Check your system clock"
-#RUN cp /src/* src/ulib/net/server/plugin/usp
+RUN cp /src/* src/ulib/net/server/plugin/usp
 RUN find . -exec touch {} \;

 RUN echo "userver {" >> $ULIB_ROOT/benchmark.cfg
--- a/tests/examples/benchmark/FrameworkBenchmarks/ULib/docker/ulib-postgres_fit.dockerfile
+++ b/tests/examples/benchmark/FrameworkBenchmarks/ULib/docker/ulib-postgres_fit.dockerfile
@ -45,7 +45,7 @@ RUN tar xf ULib-${ULIB_VERSION}.tar.gz
 WORKDIR $IROOT/ULib-$ULIB_VERSION

 # AVOID "configure: error: newly created file is older than distributed files! Check your system clock"
-#RUN cp /src/* src/ulib/net/server/plugin/usp
+RUN cp /src/* src/ulib/net/server/plugin/usp
 RUN find . -exec touch {} \;

 RUN echo "userver {" >> $ULIB_ROOT/benchmark.cfg
--- a/tests/examples/benchmark/FrameworkBenchmarks/ULib/docker/ulib.dockerfile
+++ b/tests/examples/benchmark/FrameworkBenchmarks/ULib/docker/ulib.dockerfile
@ -45,7 +45,7 @@ RUN tar xf ULib-${ULIB_VERSION}.tar.gz
 WORKDIR $IROOT/ULib-$ULIB_VERSION

 # AVOID "configure: error: newly created file is older than distributed files! Check your system clock"
-#RUN cp /src/* src/ulib/net/server/plugin/usp 
+RUN cp /src/* src/ulib/net/server/plugin/usp 
 RUN find . -exec touch {} \;

 RUN echo "userver {" >> $ULIB_ROOT/benchmark.cfg
 @ -1 +1 @@
 E