// file: staging/src/github.com/kubeedge/beehive/pkg/core/core.go
func StartModules() {
    // initialize channel-mode beehive for this process
    beehiveContext.InitContext([]string{common.MsgCtxTypeChannel})
    for name, module := range GetModules() {
        beehiveContext.AddModule(&common.ModuleInfo{ModuleName: name, ModuleType: module.contextType})
        beehiveContext.AddModuleGroup(name, module.module.Group())
        go localModuleKeeper(module)
    }
}

// file: cloud/pkg/common/messagelayer/context.go
func (cml *ContextMessageLayer) Send(message model.Message) error {
    module := cml.SendModuleName
    if len(cml.SendRouterModuleName) != 0 && isRouterMsg(message) {
        module = cml.SendRouterModuleName
    }
    beehiveContext.Send(module, message) // in-process only
    return nil
}

// file: cloud/pkg/synccontroller/objectsync.go
if CompareResourceVersion(objectRV, sync.Status.ObjectResourceVersion) > 0 {
    msg := buildEdgeControllerMessage(nodeName, sync.Namespace, resourceType,
        sync.Spec.ObjectName, model.UpdateOperation, obj)
    beehiveContext.Send(commonconst.DefaultContextSendModuleName, *msg) // to CloudHub in same process
}

// file: cloud/pkg/cloudhub/dispatcher/message_dispatcher.go
// namespaced resource path
objectSync := &v1alpha1.ObjectSync{ /* APIVersion, Kind, Name */ }
created, err := md.reliableClient.ReliablesyncsV1alpha1().
    ObjectSyncs(resourceNamespace).Create(ctx, objectSync, metav1.CreateOptions{})
created.Status.ObjectResourceVersion = "0"
_, _ = md.reliableClient.ReliablesyncsV1alpha1().
    ObjectSyncs(resourceNamespace).UpdateStatus(ctx, created, metav1.UpdateOptions{})

// non-namespaced (cluster) resource path
cos := &v1alpha1.ClusterObjectSync{ /* APIVersion, Kind, Name */ }
cos, _ = md.reliableClient.ReliablesyncsV1alpha1().
    ClusterObjectSyncs().Create(ctx, cos, metav1.CreateOptions{})
cos.Status.ObjectResourceVersion = "0"
_, _ = md.reliableClient.ReliablesyncsV1alpha1().
    ClusterObjectSyncs().UpdateStatus(ctx, cos, metav1.UpdateOptions{})

// file: cloud/pkg/cloudhub/session/node_session.go
// namespaced
objectSync.Status.ObjectResourceVersion = msg.GetResourceVersion()
_, _ = ns.reliableClient.ReliablesyncsV1alpha1().
    ObjectSyncs(resourceNamespace).UpdateStatus(ctx, objectSync, metav1.UpdateOptions{})

// cluster-scoped
clusterObjectSync.Status.ObjectResourceVersion = msg.GetResourceVersion()
_, _ = ns.reliableClient.ReliablesyncsV1alpha1().
    ClusterObjectSyncs().UpdateStatus(ctx, clusterObjectSync, metav1.UpdateOptions{})

// file: cloud/pkg/cloudhub/handler/message_handler.go
nodeMessagePool := common.InitNodeMessagePool(nodeID)
mh.MessageDispatcher.AddNodeMessagePool(nodeID, nodeMessagePool)
nodeSession := session.NewNodeSession(nodeID, projectID, connection, keepaliveInterval,
    nodeMessagePool, mh.reliableClient)
mh.SessionManager.AddSession(nodeSession)
// on disconnect:
mh.MessageDispatcher.DeleteNodeMessagePool(nodeInfo.NodeID, nodeMessagePool)
mh.SessionManager.DeleteSession(nodeSession)

// file: cloud/pkg/cloudhub/dispatcher/message_dispatcher.go
func (md *messageDispatcher) Publish(msg *beehivemodel.Message) error {
    switch msg.Router.Source {
    case metaserver.MetaServerSource:
        beehivecontext.Send(modules.DynamicControllerModuleName, *msg)
    case model.ResTwin:
        beehivecontext.SendToGroup(modules.DeviceControllerModuleGroup, *msg)
    default:
        beehivecontext.SendToGroup(modules.EdgeControllerGroupName, *msg)
    }
    return nil
}

// file: cloud/pkg/cloudhub/common/message_pool.go
type NodeMessagePool struct {
    AckMessageStore   cache.Store
    AckMessageQueue   workqueue.RateLimitingInterface
    NoAckMessageStore cache.Store
    NoAckMessageQueue workqueue.RateLimitingInterface
}