最简单的Streams框架，为什么不对？

peakzhang

//头文件

2. #include "ace/Task.h"
3. #include "ace/Synch.h"
4. #include "ace/Message_Block.h"
5. #include "ace/Log_Msg.h"
6. #include "ace/Stream.h"
7. class S_Task_1 : public ACE_Task<ACE_MT_SYNCH>
8. {
9. int open(void* = 0){return this->activate();}
10. int svc(void)
11. {
12.   ACE_Message_Block *mb;
13.   for (mb = 0; ; )
14.   {
15.    ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("S_Task::svc() - ") ACE_TEXT ("waiting for work\n" )));
16.    this->getq (mb);
17.    
18.    ACE_DEBUG((LM_DEBUG, ACE_TEXT ("S_Task_1 run \n")));
19.    ACE_DEBUG((LM_DEBUG, ACE_TEXT ("Data %s\n"), mb->rd_ptr()));
20.    this->put_next(mb);   
21.   }
22. }
23. };
24. class S_Task_2 : public ACE_Task<ACE_MT_SYNCH>
25. {
26. int open(void* = 0){return this->activate();}
27. int svc(void)
28. {
29.   ACE_Message_Block *mb;
30.   for (mb = 0; ; )
31.   {
32.    ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("S_Task::svc() - ") ACE_TEXT ("waiting for work\n" )));
33.    this->getq (mb);
34.    
35.    ACE_DEBUG((LM_DEBUG, ACE_TEXT ("S_Task_2 run \n")));
36.    ACE_DEBUG((LM_DEBUG, ACE_TEXT ("Data %s\n"), mb->rd_ptr()));
37.    this->put_next(mb);
38.   }
39. }
40. };
41. class End_Task : public ACE_Task<ACE_MT_SYNCH>
42. {
43. int open(void* = 0){return this->activate();}
44. int svc(void)
45. {
46.   ACE_Message_Block *mb;
47.   for (mb = 0; ; )
48.   {
49.    ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("S_Task::svc() - ") ACE_TEXT ("waiting for work\n" )));
50.    this->getq (mb);
51.    
52.    ACE_DEBUG((LM_DEBUG, ACE_TEXT ("End_Task run \n")));
53.    ACE_DEBUG((LM_DEBUG, ACE_TEXT ("Data %s\n"), mb->rd_ptr()));
54.    mb->release();
55.   }
56. }
57. };
58. class S_Stream : public ACE_Stream<ACE_MT_SYNCH>
59. {
60. public:
61. typedef ACE_Stream<ACE_MT_SYNCH> inherited;
62. typedef ACE_Module<ACE_MT_SYNCH> Module;
63. S_Stream():inherited(){};
64. int open (void)
65. {
66.   ACE_DEBUG((LM_DEBUG, ACE_TEXT("Stream open!\n")));
67.   Module* endModule;
68.   End_Task * end_ = new End_Task();
69.   ACE_NEW_RETURN(endModule,
70.    Module("End Module", end_),
71.    -1);
72.   this->inherited::open((void*)0, (Module*)0, endModule);
74.   Module* module1;
75.   ACE_NEW_RETURN(module1,
76.    Module("module1", new S_Task_1()),
77.    -1);
78.   Module* module2;
79.   ACE_NEW_RETURN(module2,
80.    Module("module2", new S_Task_2()),
81.    -1);
82.   this->push(module1);
83.   this->push(module2);  
84.   return 0;
85. }
86. };
87. //实现文件
88. #include "SimpleStream.h"
89. #include <stdlib.h>
90. #include <iostream>
91. using namespace std;
92. int ACE_TMAIN (int, ACE_TCHAR *[])
93. {
95. ACE_Message_Block *input = new ACE_Message_Block();
96. char *testData = "Test Data Body!";
97. input->copy(testData, strlen(testData));
98. S_Stream s;
99. s.open();
100. s.put(input);
101. ACE_Message_Block *output;
102. s.get(output);
103. cout<< "Message Body:" << *output->rd_ptr();
105. system("PAUSE");
106. return 0;
107. }

复制代码

编译完了以后几个Module都阻塞在get方法上了，我在主函数里put数据了啊，为什么所有的Module都get（）不到数据呢？？
谁来给解决一下啊，我初学Stream框架，谢谢了！！

peakzhang

你要重新定义ace_task中的open方法，这样才能把message_block压入当前的message_queue,

int put(ACE_Message_Block *message,ACE_Time_Value *timeout)

{

          return this->putq(message,timeout);

}

一定要patient呀，看完书，再写测试代码！

peakzhang

问题解决，没重载put方法，看书不仔细。。。

xinshuow

能把完整的代码贴出来吗,我加了后还不行...